diff --git a/airflow/api_connexion/endpoints/import_error_endpoint.py b/airflow/api_connexion/endpoints/import_error_endpoint.py index 76b706eac1ae4..633dd0bebde52 100644 --- a/airflow/api_connexion/endpoints/import_error_endpoint.py +++ b/airflow/api_connexion/endpoints/import_error_endpoint.py @@ -31,6 +31,7 @@ from airflow.auth.managers.models.resource_details import AccessView, DagDetails from airflow.models.dag import DagModel from airflow.models.errors import ParseImportError +from airflow.utils.api_migration import mark_fastapi_migration_done from airflow.utils.session import NEW_SESSION, provide_session from airflow.www.extensions.init_auth_manager import get_auth_manager @@ -41,6 +42,7 @@ from airflow.auth.managers.models.batch_apis import IsAuthorizedDagRequest +@mark_fastapi_migration_done @security.requires_access_view(AccessView.IMPORT_ERRORS) @provide_session def get_import_error(*, import_error_id: int, session: Session = NEW_SESSION) -> APIResponse: @@ -72,6 +74,7 @@ def get_import_error(*, import_error_id: int, session: Session = NEW_SESSION) -> return import_error_schema.dump(error) +@mark_fastapi_migration_done @security.requires_access_view(AccessView.IMPORT_ERRORS) @format_parameters({"limit": check_limit}) @provide_session diff --git a/airflow/api_fastapi/common/parameters.py b/airflow/api_fastapi/common/parameters.py index bd65017637227..218077ca59e59 100644 --- a/airflow/api_fastapi/common/parameters.py +++ b/airflow/api_fastapi/common/parameters.py @@ -32,6 +32,7 @@ from airflow.models.dag import DagModel, DagTag from airflow.models.dagrun import DagRun from airflow.models.dagwarning import DagWarning, DagWarningType +from airflow.models.errors import ParseImportError from airflow.utils import timezone from airflow.utils.state import DagRunState @@ -158,6 +159,7 @@ class SortParam(BaseParam[str]): "last_run_state": DagRun.state, "last_run_start_date": DagRun.start_date, "connection_id": Connection.conn_id, + "import_error_id": ParseImportError.id, } def __init__( diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml index e844cbceeb474..28e38884803a7 100644 --- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml +++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml @@ -1538,6 +1538,105 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /public/importErrors/{import_error_id}: + get: + tags: + - Import Error + summary: Get Import Error + description: Get an import error. + operationId: get_import_error + parameters: + - name: import_error_id + in: path + required: true + schema: + type: integer + title: Import Error Id + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ImportErrorResponse' + '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/importErrors/: + get: + tags: + - Import Error + summary: Get Import Errors + description: Get all import errors. + operationId: get_import_errors + parameters: + - name: limit + in: query + required: false + schema: + type: integer + default: 100 + title: Limit + - name: offset + in: query + required: false + schema: + type: integer + default: 0 + title: Offset + - name: order_by + in: query + required: false + schema: + type: string + default: id + title: Order By + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ImportErrorCollectionResponse' + '401': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPExceptionResponse' + description: Unauthorized + '403': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPExceptionResponse' + description: Forbidden + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /public/monitor/health: get: tags: @@ -3454,6 +3553,45 @@ components: - task_instance_states title: HistoricalMetricDataResponse description: Historical Metric Data serializer for responses. + ImportErrorCollectionResponse: + properties: + import_errors: + items: + $ref: '#/components/schemas/ImportErrorResponse' + type: array + title: Import Errors + total_entries: + type: integer + title: Total Entries + type: object + required: + - import_errors + - total_entries + title: ImportErrorCollectionResponse + description: Import Error Collection Response. + ImportErrorResponse: + properties: + import_error_id: + type: integer + title: Import Error Id + timestamp: + type: string + format: date-time + title: Timestamp + filename: + type: string + title: Filename + stack_trace: + type: string + title: Stack Trace + type: object + required: + - import_error_id + - timestamp + - filename + - stack_trace + title: ImportErrorResponse + description: Import Error Response. JobResponse: properties: id: diff --git a/airflow/api_fastapi/core_api/routes/public/__init__.py b/airflow/api_fastapi/core_api/routes/public/__init__.py index a443f5a28ae8c..68caa2d77545b 100644 --- a/airflow/api_fastapi/core_api/routes/public/__init__.py +++ b/airflow/api_fastapi/core_api/routes/public/__init__.py @@ -25,6 +25,7 @@ from airflow.api_fastapi.core_api.routes.public.dag_warning import dag_warning_router from airflow.api_fastapi.core_api.routes.public.dags import dags_router from airflow.api_fastapi.core_api.routes.public.event_logs import event_logs_router +from airflow.api_fastapi.core_api.routes.public.import_error import import_error_router from airflow.api_fastapi.core_api.routes.public.monitor import monitor_router from airflow.api_fastapi.core_api.routes.public.plugins import plugins_router from airflow.api_fastapi.core_api.routes.public.pools import pools_router @@ -43,6 +44,7 @@ public_router.include_router(dag_sources_router) public_router.include_router(dags_router) public_router.include_router(event_logs_router) +public_router.include_router(import_error_router) public_router.include_router(monitor_router) public_router.include_router(dag_warning_router) public_router.include_router(plugins_router) diff --git a/airflow/api_fastapi/core_api/routes/public/import_error.py b/airflow/api_fastapi/core_api/routes/public/import_error.py new file mode 100644 index 0000000000000..9007d6ff891b9 --- /dev/null +++ b/airflow/api_fastapi/core_api/routes/public/import_error.py @@ -0,0 +1,105 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from fastapi import Depends, HTTPException, status +from sqlalchemy import select +from sqlalchemy.orm import Session +from typing_extensions import Annotated + +from airflow.api_fastapi.common.db.common import ( + get_session, + paginated_select, +) +from airflow.api_fastapi.common.parameters import ( + QueryLimit, + QueryOffset, + SortParam, +) +from airflow.api_fastapi.common.router import AirflowRouter +from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc +from airflow.api_fastapi.core_api.serializers.import_error import ( + ImportErrorCollectionResponse, + ImportErrorResponse, +) +from airflow.models.errors import ParseImportError + +import_error_router = AirflowRouter(tags=["Import Error"], prefix="/importErrors") + + +@import_error_router.get( + "/{import_error_id}", + responses=create_openapi_http_exception_doc( + [status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND] + ), +) +async def get_import_error( + import_error_id: int, + session: Annotated[Session, Depends(get_session)], +) -> ImportErrorResponse: + """Get an import error.""" + error = session.scalar(select(ParseImportError).where(ParseImportError.id == import_error_id)) + if error is None: + raise HTTPException(404, f"The ImportError with import_error_id: `{import_error_id}` was not found") + + return ImportErrorResponse.model_validate( + error, + from_attributes=True, + ) + + +@import_error_router.get( + "/", + responses=create_openapi_http_exception_doc([status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]), +) +async def get_import_errors( + limit: QueryLimit, + offset: QueryOffset, + order_by: Annotated[ + SortParam, + Depends( + SortParam( + [ + "id", + "import_error_id", + "timestamp", + "filename", + "stacktrace", + ], + ParseImportError, + ).dynamic_depends() + ), + ], + session: Annotated[Session, Depends(get_session)], +) -> ImportErrorCollectionResponse: + """Get all import errors.""" + import_errors_select, total_entries = paginated_select( + select(ParseImportError), + [], + order_by, + offset, + limit, + session, + ) + import_errors = session.scalars(import_errors_select).all() + + return ImportErrorCollectionResponse( + import_errors=[ + ImportErrorResponse.model_validate(error, from_attributes=True) for error in import_errors + ], + total_entries=total_entries, + ) diff --git a/airflow/api_fastapi/core_api/serializers/import_error.py b/airflow/api_fastapi/core_api/serializers/import_error.py new file mode 100644 index 0000000000000..ebc65e23eccbe --- /dev/null +++ b/airflow/api_fastapi/core_api/serializers/import_error.py @@ -0,0 +1,39 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + + +class ImportErrorResponse(BaseModel): + """Import Error Response.""" + + id: int = Field(alias="import_error_id") + timestamp: datetime + filename: str + stacktrace: str = Field(alias="stack_trace") + + model_config = ConfigDict(populate_by_name=True) + + +class ImportErrorCollectionResponse(BaseModel): + """Import Error Collection Response.""" + + import_errors: list[ImportErrorResponse] + total_entries: int diff --git a/airflow/ui/openapi-gen/queries/common.ts b/airflow/ui/openapi-gen/queries/common.ts index 1248a77ce188d..36ea524e0138f 100644 --- a/airflow/ui/openapi-gen/queries/common.ts +++ b/airflow/ui/openapi-gen/queries/common.ts @@ -12,6 +12,7 @@ import { DagsService, DashboardService, EventLogService, + ImportErrorService, MonitorService, PluginService, PoolService, @@ -421,6 +422,50 @@ export const UseEventLogServiceGetEventLogsKeyFn = ( }, ]), ]; +export type ImportErrorServiceGetImportErrorDefaultResponse = Awaited< + ReturnType +>; +export type ImportErrorServiceGetImportErrorQueryResult< + TData = ImportErrorServiceGetImportErrorDefaultResponse, + TError = unknown, +> = UseQueryResult; +export const useImportErrorServiceGetImportErrorKey = + "ImportErrorServiceGetImportError"; +export const UseImportErrorServiceGetImportErrorKeyFn = ( + { + importErrorId, + }: { + importErrorId: number; + }, + queryKey?: Array, +) => [ + useImportErrorServiceGetImportErrorKey, + ...(queryKey ?? [{ importErrorId }]), +]; +export type ImportErrorServiceGetImportErrorsDefaultResponse = Awaited< + ReturnType +>; +export type ImportErrorServiceGetImportErrorsQueryResult< + TData = ImportErrorServiceGetImportErrorsDefaultResponse, + TError = unknown, +> = UseQueryResult; +export const useImportErrorServiceGetImportErrorsKey = + "ImportErrorServiceGetImportErrors"; +export const UseImportErrorServiceGetImportErrorsKeyFn = ( + { + limit, + offset, + orderBy, + }: { + limit?: number; + offset?: number; + orderBy?: string; + } = {}, + queryKey?: Array, +) => [ + useImportErrorServiceGetImportErrorsKey, + ...(queryKey ?? [{ limit, offset, orderBy }]), +]; export type MonitorServiceGetHealthDefaultResponse = Awaited< ReturnType >; diff --git a/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow/ui/openapi-gen/queries/prefetch.ts index bf6ad800be012..6c41b7a5a1b55 100644 --- a/airflow/ui/openapi-gen/queries/prefetch.ts +++ b/airflow/ui/openapi-gen/queries/prefetch.ts @@ -12,6 +12,7 @@ import { DagsService, DashboardService, EventLogService, + ImportErrorService, MonitorService, PluginService, PoolService, @@ -543,6 +544,59 @@ export const prefetchUseEventLogServiceGetEventLogs = ( tryNumber, }), }); +/** + * Get Import Error + * Get an import error. + * @param data The data for the request. + * @param data.importErrorId + * @returns ImportErrorResponse Successful Response + * @throws ApiError + */ +export const prefetchUseImportErrorServiceGetImportError = ( + queryClient: QueryClient, + { + importErrorId, + }: { + importErrorId: number; + }, +) => + queryClient.prefetchQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorKeyFn({ + importErrorId, + }), + queryFn: () => ImportErrorService.getImportError({ importErrorId }), + }); +/** + * Get Import Errors + * Get all import errors. + * @param data The data for the request. + * @param data.limit + * @param data.offset + * @param data.orderBy + * @returns ImportErrorCollectionResponse Successful Response + * @throws ApiError + */ +export const prefetchUseImportErrorServiceGetImportErrors = ( + queryClient: QueryClient, + { + limit, + offset, + orderBy, + }: { + limit?: number; + offset?: number; + orderBy?: string; + } = {}, +) => + queryClient.prefetchQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorsKeyFn({ + limit, + offset, + orderBy, + }), + queryFn: () => + ImportErrorService.getImportErrors({ limit, offset, orderBy }), + }); /** * Get Health * @returns HealthInfoSchema Successful Response diff --git a/airflow/ui/openapi-gen/queries/queries.ts b/airflow/ui/openapi-gen/queries/queries.ts index 70796be401719..f4b7c41195f77 100644 --- a/airflow/ui/openapi-gen/queries/queries.ts +++ b/airflow/ui/openapi-gen/queries/queries.ts @@ -17,6 +17,7 @@ import { DagsService, DashboardService, EventLogService, + ImportErrorService, MonitorService, PluginService, PoolService, @@ -671,6 +672,72 @@ export const useEventLogServiceGetEventLogs = < }) as TData, ...options, }); +/** + * Get Import Error + * Get an import error. + * @param data The data for the request. + * @param data.importErrorId + * @returns ImportErrorResponse Successful Response + * @throws ApiError + */ +export const useImportErrorServiceGetImportError = < + TData = Common.ImportErrorServiceGetImportErrorDefaultResponse, + TError = unknown, + TQueryKey extends Array = unknown[], +>( + { + importErrorId, + }: { + importErrorId: number; + }, + queryKey?: TQueryKey, + options?: Omit, "queryKey" | "queryFn">, +) => + useQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorKeyFn( + { importErrorId }, + queryKey, + ), + queryFn: () => + ImportErrorService.getImportError({ importErrorId }) as TData, + ...options, + }); +/** + * Get Import Errors + * Get all import errors. + * @param data The data for the request. + * @param data.limit + * @param data.offset + * @param data.orderBy + * @returns ImportErrorCollectionResponse Successful Response + * @throws ApiError + */ +export const useImportErrorServiceGetImportErrors = < + TData = Common.ImportErrorServiceGetImportErrorsDefaultResponse, + TError = unknown, + TQueryKey extends Array = unknown[], +>( + { + limit, + offset, + orderBy, + }: { + limit?: number; + offset?: number; + orderBy?: string; + } = {}, + queryKey?: TQueryKey, + options?: Omit, "queryKey" | "queryFn">, +) => + useQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorsKeyFn( + { limit, offset, orderBy }, + queryKey, + ), + queryFn: () => + ImportErrorService.getImportErrors({ limit, offset, orderBy }) as TData, + ...options, + }); /** * Get Health * @returns HealthInfoSchema Successful Response diff --git a/airflow/ui/openapi-gen/queries/suspense.ts b/airflow/ui/openapi-gen/queries/suspense.ts index 4f75c2ba0ce47..2870605672be6 100644 --- a/airflow/ui/openapi-gen/queries/suspense.ts +++ b/airflow/ui/openapi-gen/queries/suspense.ts @@ -12,6 +12,7 @@ import { DagsService, DashboardService, EventLogService, + ImportErrorService, MonitorService, PluginService, PoolService, @@ -657,6 +658,72 @@ export const useEventLogServiceGetEventLogsSuspense = < }) as TData, ...options, }); +/** + * Get Import Error + * Get an import error. + * @param data The data for the request. + * @param data.importErrorId + * @returns ImportErrorResponse Successful Response + * @throws ApiError + */ +export const useImportErrorServiceGetImportErrorSuspense = < + TData = Common.ImportErrorServiceGetImportErrorDefaultResponse, + TError = unknown, + TQueryKey extends Array = unknown[], +>( + { + importErrorId, + }: { + importErrorId: number; + }, + queryKey?: TQueryKey, + options?: Omit, "queryKey" | "queryFn">, +) => + useSuspenseQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorKeyFn( + { importErrorId }, + queryKey, + ), + queryFn: () => + ImportErrorService.getImportError({ importErrorId }) as TData, + ...options, + }); +/** + * Get Import Errors + * Get all import errors. + * @param data The data for the request. + * @param data.limit + * @param data.offset + * @param data.orderBy + * @returns ImportErrorCollectionResponse Successful Response + * @throws ApiError + */ +export const useImportErrorServiceGetImportErrorsSuspense = < + TData = Common.ImportErrorServiceGetImportErrorsDefaultResponse, + TError = unknown, + TQueryKey extends Array = unknown[], +>( + { + limit, + offset, + orderBy, + }: { + limit?: number; + offset?: number; + orderBy?: string; + } = {}, + queryKey?: TQueryKey, + options?: Omit, "queryKey" | "queryFn">, +) => + useSuspenseQuery({ + queryKey: Common.UseImportErrorServiceGetImportErrorsKeyFn( + { limit, offset, orderBy }, + queryKey, + ), + queryFn: () => + ImportErrorService.getImportErrors({ limit, offset, orderBy }) as TData, + ...options, + }); /** * Get Health * @returns HealthInfoSchema Successful Response diff --git a/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow/ui/openapi-gen/requests/schemas.gen.ts index 53272eae2e779..517743af17f6c 100644 --- a/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -1740,6 +1740,52 @@ export const $HistoricalMetricDataResponse = { description: "Historical Metric Data serializer for responses.", } as const; +export const $ImportErrorCollectionResponse = { + properties: { + import_errors: { + items: { + $ref: "#/components/schemas/ImportErrorResponse", + }, + type: "array", + title: "Import Errors", + }, + total_entries: { + type: "integer", + title: "Total Entries", + }, + }, + type: "object", + required: ["import_errors", "total_entries"], + title: "ImportErrorCollectionResponse", + description: "Import Error Collection Response.", +} as const; + +export const $ImportErrorResponse = { + properties: { + import_error_id: { + type: "integer", + title: "Import Error Id", + }, + timestamp: { + type: "string", + format: "date-time", + title: "Timestamp", + }, + filename: { + type: "string", + title: "Filename", + }, + stack_trace: { + type: "string", + title: "Stack Trace", + }, + }, + type: "object", + required: ["import_error_id", "timestamp", "filename", "stack_trace"], + title: "ImportErrorResponse", + description: "Import Error Response.", +} as const; + export const $JobResponse = { properties: { id: { diff --git a/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow/ui/openapi-gen/requests/services.gen.ts index fa5c7739c909b..5597b0a6a9cfe 100644 --- a/airflow/ui/openapi-gen/requests/services.gen.ts +++ b/airflow/ui/openapi-gen/requests/services.gen.ts @@ -53,6 +53,10 @@ import type { GetEventLogResponse, GetEventLogsData, GetEventLogsResponse, + GetImportErrorData, + GetImportErrorResponse, + GetImportErrorsData, + GetImportErrorsResponse, GetHealthResponse, ListDagWarningsData, ListDagWarningsResponse, @@ -865,6 +869,63 @@ export class EventLogService { } } +export class ImportErrorService { + /** + * Get Import Error + * Get an import error. + * @param data The data for the request. + * @param data.importErrorId + * @returns ImportErrorResponse Successful Response + * @throws ApiError + */ + public static getImportError( + data: GetImportErrorData, + ): CancelablePromise { + return __request(OpenAPI, { + method: "GET", + url: "/public/importErrors/{import_error_id}", + path: { + import_error_id: data.importErrorId, + }, + errors: { + 401: "Unauthorized", + 403: "Forbidden", + 404: "Not Found", + 422: "Validation Error", + }, + }); + } + + /** + * Get Import Errors + * Get all import errors. + * @param data The data for the request. + * @param data.limit + * @param data.offset + * @param data.orderBy + * @returns ImportErrorCollectionResponse Successful Response + * @throws ApiError + */ + public static getImportErrors( + data: GetImportErrorsData = {}, + ): CancelablePromise { + return __request(OpenAPI, { + method: "GET", + url: "/public/importErrors/", + query: { + limit: data.limit, + offset: data.offset, + order_by: data.orderBy, + }, + errors: { + 401: "Unauthorized", + 403: "Forbidden", + 422: "Validation Error", + }, + }); + } +} + export class MonitorService { /** * Get Health diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow/ui/openapi-gen/requests/types.gen.ts index 909b78dd627e1..e3071b64936a8 100644 --- a/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow/ui/openapi-gen/requests/types.gen.ts @@ -416,6 +416,24 @@ export type HistoricalMetricDataResponse = { task_instance_states: airflow__api_fastapi__core_api__serializers__dashboard__TaskInstanceState; }; +/** + * Import Error Collection Response. + */ +export type ImportErrorCollectionResponse = { + import_errors: Array; + total_entries: number; +}; + +/** + * Import Error Response. + */ +export type ImportErrorResponse = { + import_error_id: number; + timestamp: string; + filename: string; + stack_trace: string; +}; + /** * Job serializer for responses. */ @@ -878,6 +896,20 @@ export type GetEventLogsData = { export type GetEventLogsResponse = EventLogCollectionResponse; +export type GetImportErrorData = { + importErrorId: number; +}; + +export type GetImportErrorResponse = ImportErrorResponse; + +export type GetImportErrorsData = { + limit?: number; + offset?: number; + orderBy?: string; +}; + +export type GetImportErrorsResponse = ImportErrorCollectionResponse; + export type GetHealthResponse = HealthInfoSchema; export type ListDagWarningsData = { @@ -1651,6 +1683,56 @@ export type $OpenApiTs = { }; }; }; + "/public/importErrors/{import_error_id}": { + get: { + req: GetImportErrorData; + res: { + /** + * Successful Response + */ + 200: ImportErrorResponse; + /** + * Unauthorized + */ + 401: HTTPExceptionResponse; + /** + * Forbidden + */ + 403: HTTPExceptionResponse; + /** + * Not Found + */ + 404: HTTPExceptionResponse; + /** + * Validation Error + */ + 422: HTTPValidationError; + }; + }; + }; + "/public/importErrors/": { + get: { + req: GetImportErrorsData; + res: { + /** + * Successful Response + */ + 200: ImportErrorCollectionResponse; + /** + * Unauthorized + */ + 401: HTTPExceptionResponse; + /** + * Forbidden + */ + 403: HTTPExceptionResponse; + /** + * Validation Error + */ + 422: HTTPValidationError; + }; + }; + }; "/public/monitor/health": { get: { res: { diff --git a/tests/api_fastapi/core_api/routes/public/test_import_error.py b/tests/api_fastapi/core_api/routes/public/test_import_error.py new file mode 100644 index 0000000000000..4271c05b6b355 --- /dev/null +++ b/tests/api_fastapi/core_api/routes/public/test_import_error.py @@ -0,0 +1,219 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from airflow.models.errors import ParseImportError +from airflow.utils.session import provide_session + +from tests_common.test_utils.db import clear_db_import_errors + +pytestmark = pytest.mark.db_test + +FILENAME1 = "test_filename1.py" +FILENAME2 = "test_filename2.py" +FILENAME3 = "Lorem_ipsum.py" +STACKTRACE1 = "test_stacktrace1" +STACKTRACE2 = "test_stacktrace2" +STACKTRACE3 = "Lorem ipsum dolor sit amet, consectetur adipiscing elit." +TIMESTAMP1 = datetime(2024, 6, 15, 1, 0, tzinfo=timezone.utc) +TIMESTAMP2 = datetime(2024, 6, 15, 5, 0, tzinfo=timezone.utc) +TIMESTAMP3 = datetime(2024, 6, 15, 3, 0, tzinfo=timezone.utc) +IMPORT_ERROR_NON_EXISTED_ID = 9999 +IMPORT_ERROR_NON_EXISTED_KEY = "non_existed_key" + + +class TestImportErrorEndpoint: + """Common class for /public/importErrors related unit tests.""" + + @staticmethod + def _clear_db(): + clear_db_import_errors() + + @pytest.fixture(autouse=True) + @provide_session + def setup(self, session=None) -> dict[str, ParseImportError]: + """ + Setup method which is run before every test. + """ + self._clear_db() + import_error1 = ParseImportError( + filename=FILENAME1, + stacktrace=STACKTRACE1, + timestamp=TIMESTAMP1, + ) + import_error2 = ParseImportError( + filename=FILENAME2, + stacktrace=STACKTRACE2, + timestamp=TIMESTAMP2, + ) + import_error3 = ParseImportError( + filename=FILENAME3, + stacktrace=STACKTRACE3, + timestamp=TIMESTAMP3, + ) + session.add_all([import_error1, import_error2, import_error3]) + session.commit() + return {FILENAME1: import_error1, FILENAME2: import_error2, FILENAME3: import_error3} + + def teardown_method(self) -> None: + self._clear_db() + + +class TestGetImportError(TestImportErrorEndpoint): + @pytest.mark.parametrize( + "import_error_key, expected_status_code, expected_body", + [ + ( + FILENAME1, + 200, + { + "import_error_id": 1, + "timestamp": TIMESTAMP1, + "filename": FILENAME1, + "stack_trace": STACKTRACE1, + }, + ), + ( + FILENAME2, + 200, + { + "import_error_id": 2, + "timestamp": TIMESTAMP2, + "filename": FILENAME2, + "stack_trace": STACKTRACE2, + }, + ), + (IMPORT_ERROR_NON_EXISTED_KEY, 404, {}), + ], + ) + def test_get_import_error( + self, test_client, setup, import_error_key, expected_status_code, expected_body + ): + import_error: ParseImportError | None = setup.get(import_error_key) + import_error_id = import_error.id if import_error else IMPORT_ERROR_NON_EXISTED_ID + response = test_client.get(f"/public/importErrors/{import_error_id}") + assert response.status_code == expected_status_code + if expected_status_code != 200: + return + expected_json = { + "import_error_id": import_error_id, + "timestamp": expected_body["timestamp"].isoformat().replace("+00:00", "Z"), + "filename": expected_body["filename"], + "stack_trace": expected_body["stack_trace"], + } + assert response.json() == expected_json + + +class TestGetImportErrors(TestImportErrorEndpoint): + @pytest.mark.parametrize( + "query_params, expected_status_code, expected_total_entries, expected_filenames", + [ + ( + {}, + 200, + 3, + [FILENAME1, FILENAME2, FILENAME3], + ), + # offset, limit + ( + {"limit": 1, "offset": 1}, + 200, + 3, + [FILENAME2], + ), + ( + {"limit": 1, "offset": 2}, + 200, + 3, + [FILENAME3], + ), + # order_by + ( + {"order_by": "-filename"}, + 200, + 3, + [FILENAME2, FILENAME1, FILENAME3], + ), + ( + {"order_by": "timestamp"}, + 200, + 3, + [FILENAME1, FILENAME3, FILENAME2], + ), + ( + {"order_by": "import_error_id"}, + 200, + 3, + [FILENAME1, FILENAME2, FILENAME3], + ), + ( + {"order_by": "-import_error_id"}, + 200, + 3, + [FILENAME3, FILENAME2, FILENAME1], + ), + # invalid order_by + ( + {"order_by": "invalid_order_by"}, + 400, + 0, + [], + ), + # combination of query parameters + ( + {"limit": 2, "offset": 1, "order_by": "-filename"}, + 200, + 3, + [FILENAME1, FILENAME3], + ), + ( + {"limit": 1, "offset": 2, "order_by": "-filename"}, + 200, + 3, + [FILENAME3], + ), + ( + {"limit": 5, "offset": 1, "order_by": "timestamp"}, + 200, + 3, + [FILENAME3, FILENAME2], + ), + ], + ) + def test_get_import_errors( + self, + test_client, + query_params, + expected_status_code, + expected_total_entries, + expected_filenames, + ): + response = test_client.get("/public/importErrors", params=query_params) + + assert response.status_code == expected_status_code + if expected_status_code != 200: + return + + response_json = response.json() + assert response_json["total_entries"] == expected_total_entries + assert [ + import_error["filename"] for import_error in response_json["import_errors"] + ] == expected_filenames