This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6c3caa6f311 AIP-84: Migrating DELETE queued asset events for DAG to
fastAPI (#44129)
6c3caa6f311 is described below
commit 6c3caa6f311ea17df02a800d1d65788e5e549f44
Author: Amogh Desai <[email protected]>
AuthorDate: Mon Nov 18 19:27:05 2024 +0530
AIP-84: Migrating DELETE queued asset events for DAG to fastAPI (#44129)
* AIP-84: Migrating GET queued asset events for DAG to fastAPI
* fixing tests and server code
* fixing parameters
* fixing parameters
* AIP-84: Migrating delete queued asset events for DAG to fastAPI
* adding assert for count
* review comments from kalyan
---
airflow/api_connexion/endpoints/asset_endpoint.py | 1 +
.../api_fastapi/core_api/openapi/v1-generated.yaml | 53 ++++++++++++++++++++++
.../api_fastapi/core_api/routes/public/assets.py | 26 ++++++++++-
airflow/ui/openapi-gen/queries/common.ts | 3 ++
airflow/ui/openapi-gen/queries/queries.ts | 42 +++++++++++++++++
airflow/ui/openapi-gen/requests/services.gen.ts | 32 +++++++++++++
airflow/ui/openapi-gen/requests/types.gen.ts | 36 +++++++++++++++
.../core_api/routes/public/test_assets.py | 44 ++++++++++++++++++
8 files changed, 236 insertions(+), 1 deletion(-)
diff --git a/airflow/api_connexion/endpoints/asset_endpoint.py
b/airflow/api_connexion/endpoints/asset_endpoint.py
index ff47db88387..d7b1c36c684 100644
--- a/airflow/api_connexion/endpoints/asset_endpoint.py
+++ b/airflow/api_connexion/endpoints/asset_endpoint.py
@@ -251,6 +251,7 @@ def get_dag_asset_queued_events(
)
+@mark_fastapi_migration_done
@security.requires_access_asset("DELETE")
@security.requires_access_dag("GET")
@action_logging
diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
index 8678b21c9b0..e6430b3d4b2 100644
--- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
+++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
@@ -487,6 +487,59 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ delete:
+ tags:
+ - Asset
+ summary: Delete Dag Asset Queued Events
+ operationId: delete_dag_asset_queued_events
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ - name: before
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Before
+ responses:
+ '204':
+ description: Successful Response
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/public/backfills/:
get:
tags:
diff --git a/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow/api_fastapi/core_api/routes/public/assets.py
index 37fbb6e5878..714920a8e0b 100644
--- a/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -21,7 +21,7 @@ from datetime import datetime
from typing import Annotated
from fastapi import Depends, HTTPException, status
-from sqlalchemy import select
+from sqlalchemy import delete, select
from sqlalchemy.orm import Session, joinedload, subqueryload
from airflow.api_fastapi.common.db.common import get_session, paginated_select
@@ -252,3 +252,27 @@ def get_dag_asset_queued_events(
],
total_entries=total_entries,
)
+
+
+@assets_router.delete(
+ "/dags/{dag_id}/assets/queuedEvent",
+ status_code=status.HTTP_204_NO_CONTENT,
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_404_NOT_FOUND,
+ ]
+ ),
+)
+def delete_dag_asset_queued_events(
+ dag_id: str,
+ session: Annotated[Session, Depends(get_session)],
+ before: OptionalDateTimeQuery = None,
+):
+ where_clause = _generate_queued_event_where_clause(dag_id=dag_id,
before=before)
+
+ delete_statement = delete(AssetDagRunQueue).where(*where_clause)
+ result = session.execute(delete_statement)
+
+ if result.rowcount == 0:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, f"Queue event with
dag_id: `{dag_id}` was not found")
diff --git a/airflow/ui/openapi-gen/queries/common.ts
b/airflow/ui/openapi-gen/queries/common.ts
index 7b23e33f0ab..9cbd31f1d6e 100644
--- a/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow/ui/openapi-gen/queries/common.ts
@@ -1158,6 +1158,9 @@ export type PoolServicePatchPoolMutationResult = Awaited<
export type VariableServicePatchVariableMutationResult = Awaited<
ReturnType<typeof VariableService.patchVariable>
>;
+export type AssetServiceDeleteDagAssetQueuedEventsMutationResult = Awaited<
+ ReturnType<typeof AssetService.deleteDagAssetQueuedEvents>
+>;
export type ConnectionServiceDeleteConnectionMutationResult = Awaited<
ReturnType<typeof ConnectionService.deleteConnection>
>;
diff --git a/airflow/ui/openapi-gen/queries/queries.ts
b/airflow/ui/openapi-gen/queries/queries.ts
index 8ec0ea9234a..a1a95bb02d5 100644
--- a/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow/ui/openapi-gen/queries/queries.ts
@@ -2551,6 +2551,48 @@ export const useVariableServicePatchVariable = <
}) as unknown as Promise<TData>,
...options,
});
+/**
+ * Delete Dag Asset Queued Events
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.before
+ * @returns void Successful Response
+ * @throws ApiError
+ */
+export const useAssetServiceDeleteDagAssetQueuedEvents = <
+ TData = Common.AssetServiceDeleteDagAssetQueuedEventsMutationResult,
+ TError = unknown,
+ TContext = unknown,
+>(
+ options?: Omit<
+ UseMutationOptions<
+ TData,
+ TError,
+ {
+ before?: string;
+ dagId: string;
+ },
+ TContext
+ >,
+ "mutationFn"
+ >,
+) =>
+ useMutation<
+ TData,
+ TError,
+ {
+ before?: string;
+ dagId: string;
+ },
+ TContext
+ >({
+ mutationFn: ({ before, dagId }) =>
+ AssetService.deleteDagAssetQueuedEvents({
+ before,
+ dagId,
+ }) as unknown as Promise<TData>,
+ ...options,
+ });
/**
* Delete Connection
* Delete a connection entry.
diff --git a/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow/ui/openapi-gen/requests/services.gen.ts
index 361cde79fc5..06e27f22ea0 100644
--- a/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -15,6 +15,8 @@ import type {
GetAssetResponse,
GetDagAssetQueuedEventsData,
GetDagAssetQueuedEventsResponse,
+ DeleteDagAssetQueuedEventsData,
+ DeleteDagAssetQueuedEventsResponse,
HistoricalMetricsData,
HistoricalMetricsResponse,
RecentDagRunsData,
@@ -306,6 +308,36 @@ export class AssetService {
},
});
}
+
+ /**
+ * Delete Dag Asset Queued Events
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.before
+ * @returns void Successful Response
+ * @throws ApiError
+ */
+ public static deleteDagAssetQueuedEvents(
+ data: DeleteDagAssetQueuedEventsData,
+ ): CancelablePromise<DeleteDagAssetQueuedEventsResponse> {
+ return __request(OpenAPI, {
+ method: "DELETE",
+ url: "/public/dags/{dag_id}/assets/queuedEvent",
+ path: {
+ dag_id: data.dagId,
+ },
+ query: {
+ before: data.before,
+ },
+ errors: {
+ 400: "Bad Request",
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
}
export class DashboardService {
diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow/ui/openapi-gen/requests/types.gen.ts
index 948b3149c4f..6065db8c50a 100644
--- a/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1069,6 +1069,13 @@ export type GetDagAssetQueuedEventsData = {
export type GetDagAssetQueuedEventsResponse = QueuedEventCollectionResponse;
+export type DeleteDagAssetQueuedEventsData = {
+ before?: string | null;
+ dagId: string;
+};
+
+export type DeleteDagAssetQueuedEventsResponse = void;
+
export type HistoricalMetricsData = {
endDate: string;
startDate: string;
@@ -1691,6 +1698,35 @@ export type $OpenApiTs = {
422: HTTPValidationError;
};
};
+ delete: {
+ req: DeleteDagAssetQueuedEventsData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 204: void;
+ /**
+ * Bad Request
+ */
+ 400: HTTPExceptionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
};
"/ui/dashboard/historical_metrics_data": {
get: {
diff --git a/tests/api_fastapi/core_api/routes/public/test_assets.py
b/tests/api_fastapi/core_api/routes/public/test_assets.py
index 4e127ffd0c7..4a3e21035c3 100644
--- a/tests/api_fastapi/core_api/routes/public/test_assets.py
+++ b/tests/api_fastapi/core_api/routes/public/test_assets.py
@@ -523,6 +523,50 @@ class TestGetDagAssetQueuedEvents(TestQueuedEventEndpoint):
assert response.json()["detail"] == "Queue event with dag_id:
`not_exists` was not found"
+class TestDeleteDagDatasetQueuedEvents(TestQueuedEventEndpoint):
+ @pytest.mark.usefixtures("time_freezer")
+ def test_should_respond_204(self, test_client, session, create_dummy_dag):
+ dag, _ = create_dummy_dag()
+ dag_id = dag.dag_id
+ self.create_assets(session=session, num=1)
+ asset_id = 1
+ self._create_asset_dag_run_queues(dag_id, asset_id, session)
+ adrqs = session.query(AssetDagRunQueue).all()
+ assert len(adrqs) == 1
+
+ response = test_client.delete(
+ f"/public/dags/{dag_id}/assets/queuedEvent",
+ )
+
+ assert response.status_code == 204
+ adrqs = session.query(AssetDagRunQueue).all()
+ assert len(adrqs) == 0
+
+ def test_should_respond_404_invalid_dag(self, test_client):
+ dag_id = "not_exists"
+
+ response = test_client.delete(
+ f"/public/dags/{dag_id}/assets/queuedEvent",
+ )
+
+ assert response.status_code == 404
+ assert response.json()["detail"] == "Queue event with dag_id:
`not_exists` was not found"
+
+ def test_should_respond_404_valid_dag_no_adrq(self, test_client, session,
create_dummy_dag):
+ dag, _ = create_dummy_dag()
+ dag_id = dag.dag_id
+ self.create_assets(session=session, num=1)
+ adrqs = session.query(AssetDagRunQueue).all()
+ assert len(adrqs) == 0
+
+ response = test_client.delete(
+ f"/public/dags/{dag_id}/assets/queuedEvent",
+ )
+
+ assert response.status_code == 404
+ assert response.json()["detail"] == "Queue event with dag_id: `dag`
was not found"
+
+
class TestPostAssetEvents(TestAssets):
@pytest.mark.usefixtures("time_freezer")
def test_should_respond_200(self, test_client, session):
