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 58ceb32391 AIP-84 Get Import Error / Get Import Errors (#43637)
58ceb32391 is described below
commit 58ceb323916d8412dafbcd03c1e99f8c36e12faa
Author: LIU ZHE YOU <[email protected]>
AuthorDate: Tue Nov 5 22:30:38 2024 +0800
AIP-84 Get Import Error / Get Import Errors (#43637)
* AIP-84 Get Import Error
* Small adjustments
---------
Co-authored-by: pierrejeambrun <[email protected]>
---
.../endpoints/import_error_endpoint.py | 3 +
airflow/api_fastapi/common/parameters.py | 2 +
.../api_fastapi/core_api/openapi/v1-generated.yaml | 138 +++++++++++++
.../api_fastapi/core_api/routes/public/__init__.py | 2 +
.../core_api/routes/public/import_error.py | 105 ++++++++++
.../core_api/serializers/import_error.py | 39 ++++
airflow/ui/openapi-gen/queries/common.ts | 45 +++++
airflow/ui/openapi-gen/queries/prefetch.ts | 54 +++++
airflow/ui/openapi-gen/queries/queries.ts | 67 +++++++
airflow/ui/openapi-gen/queries/suspense.ts | 67 +++++++
airflow/ui/openapi-gen/requests/schemas.gen.ts | 46 +++++
airflow/ui/openapi-gen/requests/services.gen.ts | 61 ++++++
airflow/ui/openapi-gen/requests/types.gen.ts | 82 ++++++++
.../core_api/routes/public/test_import_error.py | 219 +++++++++++++++++++++
14 files changed, 930 insertions(+)
diff --git a/airflow/api_connexion/endpoints/import_error_endpoint.py
b/airflow/api_connexion/endpoints/import_error_endpoint.py
index 76b706eac1..633dd0bebd 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.api_connexion.schemas.error_schema import (
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 @@ if TYPE_CHECKING:
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 bd65017637..218077ca59 100644
--- a/airflow/api_fastapi/common/parameters.py
+++ b/airflow/api_fastapi/common/parameters.py
@@ -32,6 +32,7 @@ from airflow.models import Base, Connection
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 e844cbceeb..28e3888480 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 a443f5a28a..68caa2d775 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_sources
import dag_sources_r
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_run_router)
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 0000000000..9007d6ff89
--- /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 0000000000..ebc65e23ec
--- /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 1248a77ce1..36ea524e01 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<typeof ImportErrorService.getImportError>
+>;
+export type ImportErrorServiceGetImportErrorQueryResult<
+ TData = ImportErrorServiceGetImportErrorDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useImportErrorServiceGetImportErrorKey =
+ "ImportErrorServiceGetImportError";
+export const UseImportErrorServiceGetImportErrorKeyFn = (
+ {
+ importErrorId,
+ }: {
+ importErrorId: number;
+ },
+ queryKey?: Array<unknown>,
+) => [
+ useImportErrorServiceGetImportErrorKey,
+ ...(queryKey ?? [{ importErrorId }]),
+];
+export type ImportErrorServiceGetImportErrorsDefaultResponse = Awaited<
+ ReturnType<typeof ImportErrorService.getImportErrors>
+>;
+export type ImportErrorServiceGetImportErrorsQueryResult<
+ TData = ImportErrorServiceGetImportErrorsDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useImportErrorServiceGetImportErrorsKey =
+ "ImportErrorServiceGetImportErrors";
+export const UseImportErrorServiceGetImportErrorsKeyFn = (
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: Array<unknown>,
+) => [
+ useImportErrorServiceGetImportErrorsKey,
+ ...(queryKey ?? [{ limit, offset, orderBy }]),
+];
export type MonitorServiceGetHealthDefaultResponse = Awaited<
ReturnType<typeof MonitorService.getHealth>
>;
diff --git a/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow/ui/openapi-gen/queries/prefetch.ts
index bf6ad800be..6c41b7a5a1 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 70796be401..f4b7c41195 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> = unknown[],
+>(
+ {
+ importErrorId,
+ }: {
+ importErrorId: number;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ 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> = unknown[],
+>(
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ 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 4f75c2ba0c..2870605672 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> = unknown[],
+>(
+ {
+ importErrorId,
+ }: {
+ importErrorId: number;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ 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> = unknown[],
+>(
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ 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 53272eae2e..517743af17 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 fa5c7739c9..5597b0a6a9 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<GetImportErrorResponse> {
+ 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<GetImportErrorsResponse> {
+ 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 909b78dd62..e3071b6493 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<ImportErrorResponse>;
+ 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 0000000000..4271c05b6b
--- /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
