AIP-84 Get Event Log (#43406)

* AIP-84 Get Event Log

* fix: add error docs, refactor test case

airflow/api_connexion/endpoints/event_log_endpoint.py

@@ -31,6 +31,7 @@
 from airflow.auth.managers.models.resource_details import DagAccessEntity
 from airflow.models import Log
 from airflow.utils import timezone
+from airflow.utils.api_migration import mark_fastapi_migration_done
 from airflow.utils.db import get_query_count
 from airflow.utils.session import NEW_SESSION, provide_session
 
@@ -40,6 +41,7 @@
     from airflow.api_connexion.types import APIResponse
 
 
+@mark_fastapi_migration_done
 @security.requires_access_dag("GET", DagAccessEntity.AUDIT_LOG)
 @provide_session
 def get_event_log(*, event_log_id: int, session: Session = NEW_SESSION) -> APIResponse:

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -1506,6 +1506,50 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/VersionInfo'
+  /public/eventLogs/{event_log_id}:
+    get:
+      tags:
+      - Event Log
+      summary: Get Event Log
+      operationId: get_event_log
+      parameters:
+      - name: event_log_id
+        in: path
+        required: true
+        schema:
+          type: integer
+          title: Event Log Id
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/EventLogResponse'
+        '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'
 components:
   schemas:
     AppBuilderMenuItemResponse:
@@ -2415,6 +2459,74 @@ components:
       title: DagTagPydantic
       description: Serializable representation of the DagTag ORM SqlAlchemyModel used
         by internal API.
+    EventLogResponse:
+      properties:
+        event_log_id:
+          type: integer
+          title: Event Log Id
+        when:
+          type: string
+          format: date-time
+          title: When
+        dag_id:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Dag Id
+        task_id:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Task Id
+        run_id:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Run Id
+        map_index:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Map Index
+        try_number:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Try Number
+        event:
+          type: string
+          title: Event
+        logical_date:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Logical Date
+        owner:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Owner
+        extra:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Extra
+      type: object
+      required:
+      - event_log_id
+      - when
+      - dag_id
+      - task_id
+      - run_id
+      - map_index
+      - try_number
+      - event
+      - logical_date
+      - owner
+      - extra
+      title: EventLogResponse
+      description: Event Log Response.
     FastAPIAppResponse:
       properties:
         app:

airflow/api_fastapi/core_api/routes/public/__init__.py

@@ -21,6 +21,7 @@
 from airflow.api_fastapi.core_api.routes.public.connections import connections_router
 from airflow.api_fastapi.core_api.routes.public.dag_run import dag_run_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.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
@@ -40,3 +41,4 @@
 public_router.include_router(providers_router)
 public_router.include_router(plugins_router)
 public_router.include_router(version_router)
+public_router.include_router(event_logs_router)

airflow/api_fastapi/core_api/routes/public/event_logs.py

@@ -0,0 +1,51 @@
+# 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
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+    get_session,
+)
+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.event_logs import (
+    EventLogResponse,
+)
+from airflow.models import Log
+
+event_logs_router = AirflowRouter(tags=["Event Log"], prefix="/eventLogs")
+
+
+@event_logs_router.get(
+    "/{event_log_id}",
+    responses=create_openapi_http_exception_doc([401, 403, 404]),
+)
+async def get_event_log(
+    event_log_id: int,
+    session: Annotated[Session, Depends(get_session)],
+) -> EventLogResponse:
+    event_log = session.scalar(select(Log).where(Log.id == event_log_id))
+    if event_log is None:
+        raise HTTPException(404, f"The Event Log with id: `{event_log_id}` not found")
+    return EventLogResponse.model_validate(
+        event_log,
+        from_attributes=True,
+    )

airflow/api_fastapi/core_api/serializers/event_logs.py

@@ -0,0 +1,40 @@
+# 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 EventLogResponse(BaseModel):
+    """Event Log Response."""
+
+    id: int = Field(alias="event_log_id")
+    dttm: datetime = Field(alias="when")
+    dag_id: str | None
+    task_id: str | None
+    run_id: str | None
+    map_index: int | None
+    try_number: int | None
+    event: str
+    execution_date: datetime | None = Field(alias="logical_date")
+    owner: str | None
+    extra: str | None
+
+    model_config = ConfigDict(populate_by_name=True)

airflow/ui/openapi-gen/queries/common.ts

@@ -8,6 +8,7 @@ import {
   DagService,
   DagsService,
   DashboardService,
+  EventLogService,
   MonitorService,
   PluginService,
   PoolService,
@@ -409,6 +410,22 @@ export const UseVersionServiceGetVersionKeyFn = (queryKey?: Array<unknown>) => [
   useVersionServiceGetVersionKey,
   ...(queryKey ?? []),
 ];
+export type EventLogServiceGetEventLogDefaultResponse = Awaited<
+  ReturnType<typeof EventLogService.getEventLog>
+>;
+export type EventLogServiceGetEventLogQueryResult<
+  TData = EventLogServiceGetEventLogDefaultResponse,
+  TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useEventLogServiceGetEventLogKey = "EventLogServiceGetEventLog";
+export const UseEventLogServiceGetEventLogKeyFn = (
+  {
+    eventLogId,
+  }: {
+    eventLogId: number;
+  },
+  queryKey?: Array<unknown>,
+) => [useEventLogServiceGetEventLogKey, ...(queryKey ?? [{ eventLogId }])];
 export type VariableServicePostVariableMutationResult = Awaited<
   ReturnType<typeof VariableService.postVariable>
 >;

airflow/ui/openapi-gen/queries/prefetch.ts

@@ -8,6 +8,7 @@ import {
   DagService,
   DagsService,
   DashboardService,
+  EventLogService,
   MonitorService,
   PluginService,
   PoolService,
@@ -512,3 +513,22 @@ export const prefetchUseVersionServiceGetVersion = (queryClient: QueryClient) =>
     queryKey: Common.UseVersionServiceGetVersionKeyFn(),
     queryFn: () => VersionService.getVersion(),
   });
+/**
+ * Get Event Log
+ * @param data The data for the request.
+ * @param data.eventLogId
+ * @returns EventLogResponse Successful Response
+ * @throws ApiError
+ */
+export const prefetchUseEventLogServiceGetEventLog = (
+  queryClient: QueryClient,
+  {
+    eventLogId,
+  }: {
+    eventLogId: number;
+  },
+) =>
+  queryClient.prefetchQuery({
+    queryKey: Common.UseEventLogServiceGetEventLogKeyFn({ eventLogId }),
+    queryFn: () => EventLogService.getEventLog({ eventLogId }),
+  });

airflow/ui/openapi-gen/queries/queries.ts

@@ -13,6 +13,7 @@ import {
   DagService,
   DagsService,
   DashboardService,
+  EventLogService,
   MonitorService,
   PluginService,
   PoolService,
@@ -663,6 +664,34 @@ export const useVersionServiceGetVersion = <
     queryFn: () => VersionService.getVersion() as TData,
     ...options,
   });
+/**
+ * Get Event Log
+ * @param data The data for the request.
+ * @param data.eventLogId
+ * @returns EventLogResponse Successful Response
+ * @throws ApiError
+ */
+export const useEventLogServiceGetEventLog = <
+  TData = Common.EventLogServiceGetEventLogDefaultResponse,
+  TError = unknown,
+  TQueryKey extends Array<unknown> = unknown[],
+>(
+  {
+    eventLogId,
+  }: {
+    eventLogId: number;
+  },
+  queryKey?: TQueryKey,
+  options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+  useQuery<TData, TError>({
+    queryKey: Common.UseEventLogServiceGetEventLogKeyFn(
+      { eventLogId },
+      queryKey,
+    ),
+    queryFn: () => EventLogService.getEventLog({ eventLogId }) as TData,
+    ...options,
+  });
 /**
  * Post Variable
  * Create a variable.

airflow/ui/openapi-gen/queries/suspense.ts

@@ -8,6 +8,7 @@ import {
   DagService,
   DagsService,
   DashboardService,
+  EventLogService,
   MonitorService,
   PluginService,
   PoolService,
@@ -652,3 +653,31 @@ export const useVersionServiceGetVersionSuspense = <
     queryFn: () => VersionService.getVersion() as TData,
     ...options,
   });
+/**
+ * Get Event Log
+ * @param data The data for the request.
+ * @param data.eventLogId
+ * @returns EventLogResponse Successful Response
+ * @throws ApiError
+ */
+export const useEventLogServiceGetEventLogSuspense = <
+  TData = Common.EventLogServiceGetEventLogDefaultResponse,
+  TError = unknown,
+  TQueryKey extends Array<unknown> = unknown[],
+>(
+  {
+    eventLogId,
+  }: {
+    eventLogId: number;
+  },
+  queryKey?: TQueryKey,
+  options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+  useSuspenseQuery<TData, TError>({
+    queryKey: Common.UseEventLogServiceGetEventLogKeyFn(
+      { eventLogId },
+      queryKey,
+    ),
+    queryFn: () => EventLogService.getEventLog({ eventLogId }) as TData,
+    ...options,
+  });