add dag_id filters (apache#47730)

Author: rawwar

airflow/api_fastapi/core_api/routes/public/assets.py

@@ -54,6 +54,7 @@
 from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import (
+    ReadableDagsFilterDep,
     requires_access_asset,
     requires_access_asset_alias,
     requires_access_dag,
@@ -81,6 +82,7 @@ def _generate_queued_event_where_clause(
     asset_id: int | None = None,
     dag_id: str | None = None,
     before: datetime | str | None = None,
+    permitted_dag_ids: set[str] | None = None,
 ) -> list:
     """Get AssetDagRunQueue where clause."""
     where_clause = []
@@ -90,6 +92,8 @@ def _generate_queued_event_where_clause(
         where_clause.append(AssetDagRunQueue.asset_id == asset_id)
     if before is not None:
         where_clause.append(AssetDagRunQueue.created_at < before)
+    if permitted_dag_ids is not None:
+        where_clause.append(AssetDagRunQueue.target_dag_id.in_(permitted_dag_ids))
     return where_clause
 
 
@@ -322,11 +326,14 @@ def materialize_asset(
 )
 def get_asset_queued_events(
     asset_id: int,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ) -> QueuedEventCollectionResponse:
     """Get queued asset events for an asset."""
-    where_clause = _generate_queued_event_where_clause(asset_id=asset_id, before=before)
+    where_clause = _generate_queued_event_where_clause(
+        asset_id=asset_id, before=before, permitted_dag_ids=readable_dags_filter.value
+    )
     query = select(AssetDagRunQueue).where(*where_clause)
 
     dag_asset_queued_events_select, total_entries = paginated_select(statement=query)
@@ -381,11 +388,14 @@ def get_asset(
 )
 def get_dag_asset_queued_events(
     dag_id: str,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ) -> QueuedEventCollectionResponse:
     """Get queued asset events for a DAG."""
-    where_clause = _generate_queued_event_where_clause(dag_id=dag_id, before=before)
+    where_clause = _generate_queued_event_where_clause(
+        dag_id=dag_id, before=before, permitted_dag_ids=readable_dags_filter.value
+    )
     query = select(AssetDagRunQueue).where(*where_clause)
 
     dag_asset_queued_events_select, total_entries = paginated_select(statement=query)
@@ -412,11 +422,14 @@ def get_dag_asset_queued_events(
 def get_dag_asset_queued_event(
     dag_id: str,
     asset_id: int,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ) -> QueuedEventResponse:
     """Get a queued asset event for a DAG."""
-    where_clause = _generate_queued_event_where_clause(dag_id=dag_id, asset_id=asset_id, before=before)
+    where_clause = _generate_queued_event_where_clause(
+        dag_id=dag_id, asset_id=asset_id, before=before, permitted_dag_ids=readable_dags_filter.value
+    )
     query = select(AssetDagRunQueue).where(*where_clause)
     adrq = session.scalar(query)
     if not adrq:
@@ -440,11 +453,14 @@ def get_dag_asset_queued_event(
 )
 def delete_asset_queued_events(
     asset_id: int,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ):
     """Delete queued asset events for an asset."""
-    where_clause = _generate_queued_event_where_clause(asset_id=asset_id, before=before)
+    where_clause = _generate_queued_event_where_clause(
+        asset_id=asset_id, before=before, permitted_dag_ids=readable_dags_filter.value
+    )
     delete_stmt = delete(AssetDagRunQueue).where(*where_clause).execution_options(synchronize_session="fetch")
     result = session.execute(delete_stmt)
     if result.rowcount == 0:
@@ -471,10 +487,13 @@ def delete_asset_queued_events(
 )
 def delete_dag_asset_queued_events(
     dag_id: str,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ):
-    where_clause = _generate_queued_event_where_clause(dag_id=dag_id, before=before)
+    where_clause = _generate_queued_event_where_clause(
+        dag_id=dag_id, before=before, permitted_dag_ids=readable_dags_filter.value
+    )
 
     delete_statement = delete(AssetDagRunQueue).where(*where_clause)
     result = session.execute(delete_statement)
@@ -501,11 +520,14 @@ def delete_dag_asset_queued_events(
 def delete_dag_asset_queued_event(
     dag_id: str,
     asset_id: int,
+    readable_dags_filter: ReadableDagsFilterDep,
     session: SessionDep,
     before: OptionalDateTimeQuery = None,
 ):
     """Delete a queued asset event for a DAG."""
-    where_clause = _generate_queued_event_where_clause(dag_id=dag_id, before=before, asset_id=asset_id)
+    where_clause = _generate_queued_event_where_clause(
+        dag_id=dag_id, before=before, asset_id=asset_id, permitted_dag_ids=readable_dags_filter.value
+    )
     delete_statement = (
         delete(AssetDagRunQueue).where(*where_clause).execution_options(synchronize_session="fetch")
     )