AIP-84 Get Import Error / Get Import Errors

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

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__(

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:

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)

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,
+    )

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