Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

AIP-84: Migrating DELETE queued asset events for assets to FASTAPI #44138

Merged
1 change: 1 addition & 0 deletions airflow/api_connexion/endpoints/asset_endpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,7 @@ def get_asset_queued_events(
)


@mark_fastapi_migration_done
@security.requires_access_asset("DELETE")
@action_logging
@provide_session
Expand Down
49 changes: 49 additions & 0 deletions airflow/api_fastapi/core_api/openapi/v1-generated.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -546,6 +546,55 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
/public/assets/queuedEvent/{uri}:
delete:
tags:
- Asset
summary: Delete Asset Queued Events
description: Delete queued asset events for an asset.
operationId: delete_asset_queued_events
parameters:
- name: uri
in: path
required: true
schema:
type: string
title: Uri
- 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
'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:
Expand Down
22 changes: 22 additions & 0 deletions airflow/api_fastapi/core_api/routes/public/assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,28 @@ def get_dag_asset_queued_events(
)


@assets_router.delete(
"/assets/queuedEvent/{uri:path}",
status_code=status.HTTP_204_NO_CONTENT,
responses=create_openapi_http_exception_doc(
[
status.HTTP_404_NOT_FOUND,
]
),
)
def delete_asset_queued_events(
uri: str,
session: Annotated[Session, Depends(get_session)],
before: OptionalDateTimeQuery = None,
):
"""Delete queued asset events for an asset."""
where_clause = _generate_queued_event_where_clause(uri=uri, before=before)
delete_stmt = delete(AssetDagRunQueue).where(*where_clause).execution_options(synchronize_session="fetch")
result = session.execute(delete_stmt)
if result.rowcount == 0:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"Queue event with uri: `{uri}` was not found")


@assets_router.delete(
"/dags/{dag_id}/assets/queuedEvent",
status_code=status.HTTP_204_NO_CONTENT,
Expand Down
3 changes: 3 additions & 0 deletions airflow/ui/openapi-gen/queries/common.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1161,6 +1161,9 @@ export type VariableServicePatchVariableMutationResult = Awaited<
export type AssetServiceDeleteDagAssetQueuedEventsMutationResult = Awaited<
ReturnType<typeof AssetService.deleteDagAssetQueuedEvents>
>;
export type AssetServiceDeleteAssetQueuedEventsMutationResult = Awaited<
ReturnType<typeof AssetService.deleteAssetQueuedEvents>
>;
export type ConnectionServiceDeleteConnectionMutationResult = Awaited<
ReturnType<typeof ConnectionService.deleteConnection>
>;
Expand Down
43 changes: 43 additions & 0 deletions airflow/ui/openapi-gen/queries/queries.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2593,6 +2593,49 @@ export const useAssetServiceDeleteDagAssetQueuedEvents = <
}) as unknown as Promise<TData>,
...options,
});
/**
* Delete Asset Queued Events
* Delete queued asset events for an asset.
* @param data The data for the request.
* @param data.uri
* @param data.before
* @returns void Successful Response
* @throws ApiError
*/
export const useAssetServiceDeleteAssetQueuedEvents = <
TData = Common.AssetServiceDeleteAssetQueuedEventsMutationResult,
TError = unknown,
TContext = unknown,
>(
options?: Omit<
UseMutationOptions<
TData,
TError,
{
before?: string;
uri: string;
},
TContext
>,
"mutationFn"
>,
) =>
useMutation<
TData,
TError,
{
before?: string;
uri: string;
},
TContext
>({
mutationFn: ({ before, uri }) =>
AssetService.deleteAssetQueuedEvents({
before,
uri,
}) as unknown as Promise<TData>,
...options,
});
/**
* Delete Connection
* Delete a connection entry.
Expand Down
32 changes: 32 additions & 0 deletions airflow/ui/openapi-gen/requests/services.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ import type {
GetDagAssetQueuedEventsResponse,
DeleteDagAssetQueuedEventsData,
DeleteDagAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData,
DeleteAssetQueuedEventsResponse,
HistoricalMetricsData,
HistoricalMetricsResponse,
RecentDagRunsData,
Expand Down Expand Up @@ -338,6 +340,36 @@ export class AssetService {
},
});
}

/**
* Delete Asset Queued Events
* Delete queued asset events for an asset.
* @param data The data for the request.
* @param data.uri
* @param data.before
* @returns void Successful Response
* @throws ApiError
*/
public static deleteAssetQueuedEvents(
data: DeleteAssetQueuedEventsData,
): CancelablePromise<DeleteAssetQueuedEventsResponse> {
return __request(OpenAPI, {
method: "DELETE",
url: "/public/assets/queuedEvent/{uri}",
path: {
uri: data.uri,
},
query: {
before: data.before,
},
errors: {
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
422: "Validation Error",
},
});
}
}

export class DashboardService {
Expand Down
34 changes: 34 additions & 0 deletions airflow/ui/openapi-gen/requests/types.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1076,6 +1076,13 @@ export type DeleteDagAssetQueuedEventsData = {

export type DeleteDagAssetQueuedEventsResponse = void;

export type DeleteAssetQueuedEventsData = {
before?: string | null;
uri: string;
};

export type DeleteAssetQueuedEventsResponse = void;

export type HistoricalMetricsData = {
endDate: string;
startDate: string;
Expand Down Expand Up @@ -1728,6 +1735,33 @@ export type $OpenApiTs = {
};
};
};
"/public/assets/queuedEvent/{uri}": {
delete: {
req: DeleteAssetQueuedEventsData;
res: {
/**
* Successful Response
*/
204: void;
/**
* Unauthorized
*/
401: HTTPExceptionResponse;
/**
* Forbidden
*/
403: HTTPExceptionResponse;
/**
* Not Found
*/
404: HTTPExceptionResponse;
/**
* Validation Error
*/
422: HTTPValidationError;
};
};
};
"/ui/dashboard/historical_metrics_data": {
get: {
req: HistoricalMetricsData;
Expand Down
27 changes: 27 additions & 0 deletions tests/api_fastapi/core_api/routes/public/test_assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -772,6 +772,33 @@ def test_invalid_attr_not_allowed(self, test_client, session):

assert response.status_code == 422


class TestDeleteAssetQueuedEvents(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
uri = "s3://bucket/key/1"
self.create_assets(session=session, num=1)
asset_id = 1
self._create_asset_dag_run_queues(dag_id, asset_id, session)

response = test_client.delete(
f"/public/assets/queuedEvent/{uri}",
pierrejeambrun marked this conversation as resolved.
Show resolved Hide resolved
)
assert response.status_code == 204
assert session.query(AssetDagRunQueue).filter_by(asset_id=1).first() is None

def test_should_respond_404(self, test_client):
uri = "not_exists"

response = test_client.delete(
f"/public/assets/queuedEvent/{uri}",
)

assert response.status_code == 404
assert response.json()["detail"] == "Queue event with uri: `not_exists` was not found"

@pytest.mark.usefixtures("time_freezer")
@pytest.mark.enable_redact
def test_should_mask_sensitive_extra(self, test_client, session):
Expand Down