This is an automated email from the ASF dual-hosted git repository.
bbovenzi 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 19a3cba6de7 Add Calendar Deadlines API Endpoint and Data Models
(#68604)
19a3cba6de7 is described below
commit 19a3cba6de7dc86c5937d6f2332f92a0dfea0aa8
Author: Richard <[email protected]>
AuthorDate: Tue Jul 7 07:29:27 2026 -0700
Add Calendar Deadlines API Endpoint and Data Models (#68604)
- Introduced new data models: CalendarDeadlineResponse and
CalendarDeadlineCollectionResponse for handling aggregated deadline counts.
- Implemented the GET /ui/calendar/{dag_id}/deadlines endpoint to retrieve
deadline data, including parameters for granularity and deadline time filters.
- Updated CalendarService to include a method for fetching deadline
calendar data, aggregating results by time bucket and missed status.
- Added unit tests for the new endpoint to ensure correct functionality and
response structure.
---
.../api_fastapi/core_api/datamodels/ui/calendar.py | 15 ++
.../api_fastapi/core_api/openapi/_private_ui.yaml | 113 ++++++++++++
.../api_fastapi/core_api/routes/ui/calendar.py | 34 +++-
.../api_fastapi/core_api/services/ui/calendar.py | 55 ++++++
.../src/airflow/ui/openapi-gen/queries/common.ts | 11 ++
.../ui/openapi-gen/queries/ensureQueryData.ts | 21 +++
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 21 +++
.../src/airflow/ui/openapi-gen/queries/queries.ts | 21 +++
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 21 +++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 42 +++++
.../ui/openapi-gen/requests/services.gen.ts | 35 +++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 43 +++++
.../core_api/routes/ui/test_calendar.py | 202 ++++++++++++++++++++-
13 files changed, 631 insertions(+), 3 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/calendar.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/calendar.py
index 9bd0359463c..09f5062a497 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/calendar.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/calendar.py
@@ -42,3 +42,18 @@ class CalendarTimeRangeCollectionResponse(BaseModel):
total_entries: int
dag_runs: list[CalendarTimeRangeResponse]
+
+
+class CalendarDeadlineResponse(BaseModel):
+ """Represents aggregated deadline counts for a specific calendar time
bucket."""
+
+ date: datetime
+ missed: bool
+ count: int
+
+
+class CalendarDeadlineCollectionResponse(BaseModel):
+ """Response model for calendar deadline aggregation results."""
+
+ total_entries: int
+ deadlines: list[CalendarDeadlineResponse]
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
index 887aee12c9b..173f231a098 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
+++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
@@ -1732,6 +1732,83 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
+ /ui/calendar/{dag_id}/deadlines:
+ get:
+ tags:
+ - Calendar
+ summary: Get Calendar Deadlines
+ description: Get aggregated deadline counts for a Dag, bucketed by
deadline_time
+ and missed status.
+ operationId: get_calendar_deadlines
+ security:
+ - OAuth2PasswordBearer: []
+ - HTTPBearer: []
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ - name: granularity
+ in: query
+ required: false
+ schema:
+ enum:
+ - hourly
+ - daily
+ type: string
+ default: daily
+ title: Granularity
+ - name: deadline_time_gte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Deadline Time Gte
+ - name: deadline_time_gt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Deadline Time Gt
+ - name: deadline_time_lte
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Deadline Time Lte
+ - name: deadline_time_lt
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
+ title: Deadline Time Lt
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/CalendarDeadlineCollectionResponse'
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/ui/teams:
get:
tags:
@@ -2064,6 +2141,42 @@ components:
- type
title: BaseNodeResponse
description: Base Node serializer for responses.
+ CalendarDeadlineCollectionResponse:
+ properties:
+ total_entries:
+ type: integer
+ title: Total Entries
+ deadlines:
+ items:
+ $ref: '#/components/schemas/CalendarDeadlineResponse'
+ type: array
+ title: Deadlines
+ type: object
+ required:
+ - total_entries
+ - deadlines
+ title: CalendarDeadlineCollectionResponse
+ description: Response model for calendar deadline aggregation results.
+ CalendarDeadlineResponse:
+ properties:
+ date:
+ type: string
+ format: date-time
+ title: Date
+ missed:
+ type: boolean
+ title: Missed
+ count:
+ type: integer
+ title: Count
+ type: object
+ required:
+ - date
+ - missed
+ - count
+ title: CalendarDeadlineResponse
+ description: Represents aggregated deadline counts for a specific
calendar time
+ bucket.
CalendarTimeRangeCollectionResponse:
properties:
total_entries:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/calendar.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/calendar.py
index 4f3746b669b..066a4845b82 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/calendar.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/calendar.py
@@ -25,10 +25,14 @@ from airflow.api_fastapi.common.dagbag import DagBagDep,
get_latest_version_of_d
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.parameters import RangeFilter,
datetime_range_filter_factory
from airflow.api_fastapi.common.router import AirflowRouter
-from airflow.api_fastapi.core_api.datamodels.ui.calendar import
CalendarTimeRangeCollectionResponse
+from airflow.api_fastapi.core_api.datamodels.ui.calendar import (
+ CalendarDeadlineCollectionResponse,
+ CalendarTimeRangeCollectionResponse,
+)
from airflow.api_fastapi.core_api.security import requires_access_dag
from airflow.api_fastapi.core_api.services.ui.calendar import CalendarService
from airflow.models.dagrun import DagRun
+from airflow.models.deadline import Deadline
calendar_router = AirflowRouter(prefix="/calendar", tags=["Calendar"])
@@ -71,3 +75,31 @@ def get_calendar(
partition_date=partition_date,
granularity=granularity,
)
+
+
+@calendar_router.get(
+ "/{dag_id}/deadlines",
+ dependencies=[
+ Depends(
+ requires_access_dag(
+ method="GET",
+ access_entity=DagAccessEntity.RUN,
+ )
+ ),
+ ],
+)
+def get_calendar_deadlines(
+ dag_id: str,
+ session: SessionDep,
+ deadline_time: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("deadline_time", Deadline))],
+ granularity: Literal["hourly", "daily"] = "daily",
+) -> CalendarDeadlineCollectionResponse:
+ """Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status."""
+ calendar_service = CalendarService()
+
+ return calendar_service.get_deadline_calendar_data(
+ dag_id=dag_id,
+ session=session,
+ deadline_time=deadline_time,
+ granularity=granularity,
+ )
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py
index ebd5a38453e..ea4f33f9149 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py
@@ -30,10 +30,13 @@ from sqlalchemy.orm import InstrumentedAttribute, Session
from airflow._shared.timezones import timezone
from airflow.api_fastapi.common.parameters import RangeFilter
from airflow.api_fastapi.core_api.datamodels.ui.calendar import (
+ CalendarDeadlineCollectionResponse,
+ CalendarDeadlineResponse,
CalendarTimeRangeCollectionResponse,
CalendarTimeRangeResponse,
)
from airflow.models.dagrun import DagRun
+from airflow.models.deadline import Deadline
from airflow.serialization.definitions.dag import SerializedDAG
from airflow.timetables._cron import CronMixin
from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
@@ -357,3 +360,55 @@ class CalendarService:
return False
return True
+
+ def get_deadline_calendar_data(
+ self,
+ dag_id: str,
+ deadline_time: RangeFilter,
+ granularity: Literal["hourly", "daily"] = "daily",
+ *,
+ session: Session,
+ ) -> CalendarDeadlineCollectionResponse:
+ """
+ Get deadline calendar data for a Dag, aggregated by time bucket and
missed status.
+
+ Args:
+ dag_id: The Dag ID
+ session: Database session
+ deadline_time: Date range filter for deadline_time
+ granularity: Time granularity ("hourly" or "daily")
+
+ Returns:
+ Aggregated deadline counts per time bucket, split by
missed/pending status
+ """
+ dialect = get_dialect_name(session)
+ time_expression =
self._get_time_truncation_expression(Deadline.deadline_time, granularity,
dialect)
+
+ select_stmt = (
+ sa.select(
+ time_expression.label("datetime"),
+ Deadline.missed,
+ sa.func.count("*").label("count"),
+ )
+ .join(Deadline.dagrun)
+ .where(DagRun.dag_id == dag_id)
+ .group_by(time_expression, Deadline.missed)
+ .order_by(time_expression.asc(), Deadline.missed.asc())
+ )
+
+ select_stmt = deadline_time.to_orm(select_stmt)
+ rows = session.execute(select_stmt).all()
+
+ results = [
+ CalendarDeadlineResponse(
+ date=row.datetime,
+ missed=row.missed,
+ count=int(row._mapping["count"]),
+ )
+ for row in rows
+ ]
+
+ return CalendarDeadlineCollectionResponse(
+ total_entries=len(results),
+ deadlines=results,
+ )
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index a34ca6b0c68..9b07c3c8649 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -1035,6 +1035,17 @@ export const UseCalendarServiceGetCalendarKeyFn = ({
dagId, granularity, logical
partitionDateLt?: string;
partitionDateLte?: string;
}, queryKey?: Array<unknown>) => [useCalendarServiceGetCalendarKey,
...(queryKey ?? [{ dagId, granularity, logicalDateGt, logicalDateGte,
logicalDateLt, logicalDateLte, partitionDateGt, partitionDateGte,
partitionDateLt, partitionDateLte }])];
+export type CalendarServiceGetCalendarDeadlinesDefaultResponse =
Awaited<ReturnType<typeof CalendarService.getCalendarDeadlines>>;
+export type CalendarServiceGetCalendarDeadlinesQueryResult<TData =
CalendarServiceGetCalendarDeadlinesDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
+export const useCalendarServiceGetCalendarDeadlinesKey =
"CalendarServiceGetCalendarDeadlines";
+export const UseCalendarServiceGetCalendarDeadlinesKeyFn = ({ dagId,
deadlineTimeGt, deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity
}: {
+ dagId: string;
+ deadlineTimeGt?: string;
+ deadlineTimeGte?: string;
+ deadlineTimeLt?: string;
+ deadlineTimeLte?: string;
+ granularity?: "hourly" | "daily";
+}, queryKey?: Array<unknown>) => [useCalendarServiceGetCalendarDeadlinesKey,
...(queryKey ?? [{ dagId, deadlineTimeGt, deadlineTimeGte, deadlineTimeLt,
deadlineTimeLte, granularity }])];
export type TeamsServiceListTeamsDefaultResponse = Awaited<ReturnType<typeof
TeamsService.listTeams>>;
export type TeamsServiceListTeamsQueryResult<TData =
TeamsServiceListTeamsDefaultResponse, TError = unknown> = UseQueryResult<TData,
TError>;
export const useTeamsServiceListTeamsKey = "TeamsServiceListTeams";
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index c4c043c93b4..501d6158503 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -2086,6 +2086,27 @@ export const ensureUseCalendarServiceGetCalendarData =
(queryClient: QueryClient
partitionDateLte?: string;
}) => queryClient.ensureQueryData({ queryKey:
Common.UseCalendarServiceGetCalendarKeyFn({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }), queryFn: () =>
CalendarService.getCalendar({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }) });
/**
+* Get Calendar Deadlines
+* Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.granularity
+* @param data.deadlineTimeGte
+* @param data.deadlineTimeGt
+* @param data.deadlineTimeLte
+* @param data.deadlineTimeLt
+* @returns CalendarDeadlineCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const ensureUseCalendarServiceGetCalendarDeadlinesData = (queryClient:
QueryClient, { dagId, deadlineTimeGt, deadlineTimeGte, deadlineTimeLt,
deadlineTimeLte, granularity }: {
+ dagId: string;
+ deadlineTimeGt?: string;
+ deadlineTimeGte?: string;
+ deadlineTimeLt?: string;
+ deadlineTimeLte?: string;
+ granularity?: "hourly" | "daily";
+}) => queryClient.ensureQueryData({ queryKey:
Common.UseCalendarServiceGetCalendarDeadlinesKeyFn({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }), queryFn: ()
=> CalendarService.getCalendarDeadlines({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }) });
+/**
* List Teams
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index 7aebe21755c..d4ade79763f 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -2086,6 +2086,27 @@ export const prefetchUseCalendarServiceGetCalendar =
(queryClient: QueryClient,
partitionDateLte?: string;
}) => queryClient.prefetchQuery({ queryKey:
Common.UseCalendarServiceGetCalendarKeyFn({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }), queryFn: () =>
CalendarService.getCalendar({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }) });
/**
+* Get Calendar Deadlines
+* Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.granularity
+* @param data.deadlineTimeGte
+* @param data.deadlineTimeGt
+* @param data.deadlineTimeLte
+* @param data.deadlineTimeLt
+* @returns CalendarDeadlineCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const prefetchUseCalendarServiceGetCalendarDeadlines = (queryClient:
QueryClient, { dagId, deadlineTimeGt, deadlineTimeGte, deadlineTimeLt,
deadlineTimeLte, granularity }: {
+ dagId: string;
+ deadlineTimeGt?: string;
+ deadlineTimeGte?: string;
+ deadlineTimeLt?: string;
+ deadlineTimeLte?: string;
+ granularity?: "hourly" | "daily";
+}) => queryClient.prefetchQuery({ queryKey:
Common.UseCalendarServiceGetCalendarDeadlinesKeyFn({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }), queryFn: ()
=> CalendarService.getCalendarDeadlines({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }) });
+/**
* List Teams
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index 42eed251fa4..a8cd6066357 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -2086,6 +2086,27 @@ export const useCalendarServiceGetCalendar = <TData =
Common.CalendarServiceGetC
partitionDateLte?: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseCalendarServiceGetCalendarKeyFn({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }, queryKey), queryFn: ()
=> CalendarService.getCalendar({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt, par [...]
/**
+* Get Calendar Deadlines
+* Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.granularity
+* @param data.deadlineTimeGte
+* @param data.deadlineTimeGt
+* @param data.deadlineTimeLte
+* @param data.deadlineTimeLt
+* @returns CalendarDeadlineCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useCalendarServiceGetCalendarDeadlines = <TData =
Common.CalendarServiceGetCalendarDeadlinesDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }: {
+ dagId: string;
+ deadlineTimeGt?: string;
+ deadlineTimeGte?: string;
+ deadlineTimeLt?: string;
+ deadlineTimeLte?: string;
+ granularity?: "hourly" | "daily";
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseCalendarServiceGetCalendarDeadlinesKeyFn({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }, queryKey),
queryFn: () => CalendarService.getCalendarDeadlines({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }) as TData,
...options });
+/**
* List Teams
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index b3f2d6f4e9f..384cfb52b01 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -2086,6 +2086,27 @@ export const useCalendarServiceGetCalendarSuspense =
<TData = Common.CalendarSer
partitionDateLte?: string;
}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseCalendarServiceGetCalendarKeyFn({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDateGt,
partitionDateGte, partitionDateLt, partitionDateLte }, queryKey), queryFn: ()
=> CalendarService.getCalendar({ dagId, granularity, logicalDateGt,
logicalDateGte, logicalDateLt, logicalDateLte, partitionDat [...]
/**
+* Get Calendar Deadlines
+* Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status.
+* @param data The data for the request.
+* @param data.dagId
+* @param data.granularity
+* @param data.deadlineTimeGte
+* @param data.deadlineTimeGt
+* @param data.deadlineTimeLte
+* @param data.deadlineTimeLt
+* @returns CalendarDeadlineCollectionResponse Successful Response
+* @throws ApiError
+*/
+export const useCalendarServiceGetCalendarDeadlinesSuspense = <TData =
Common.CalendarServiceGetCalendarDeadlinesDefaultResponse, TError = unknown,
TQueryKey extends Array<unknown> = unknown[]>({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }: {
+ dagId: string;
+ deadlineTimeGt?: string;
+ deadlineTimeGte?: string;
+ deadlineTimeLt?: string;
+ deadlineTimeLte?: string;
+ granularity?: "hourly" | "daily";
+}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseCalendarServiceGetCalendarDeadlinesKeyFn({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }, queryKey),
queryFn: () => CalendarService.getCalendarDeadlines({ dagId, deadlineTimeGt,
deadlineTimeGte, deadlineTimeLt, deadlineTimeLte, granularity }) as TData,
...options });
+/**
* List Teams
* @param data The data for the request.
* @param data.limit
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 4dab4ff145c..ec5024d981e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -8581,6 +8581,48 @@ export const $BaseNodeResponse = {
description: 'Base Node serializer for responses.'
} as const;
+export const $CalendarDeadlineCollectionResponse = {
+ properties: {
+ total_entries: {
+ type: 'integer',
+ title: 'Total Entries'
+ },
+ deadlines: {
+ items: {
+ '$ref': '#/components/schemas/CalendarDeadlineResponse'
+ },
+ type: 'array',
+ title: 'Deadlines'
+ }
+ },
+ type: 'object',
+ required: ['total_entries', 'deadlines'],
+ title: 'CalendarDeadlineCollectionResponse',
+ description: 'Response model for calendar deadline aggregation results.'
+} as const;
+
+export const $CalendarDeadlineResponse = {
+ properties: {
+ date: {
+ type: 'string',
+ format: 'date-time',
+ title: 'Date'
+ },
+ missed: {
+ type: 'boolean',
+ title: 'Missed'
+ },
+ count: {
+ type: 'integer',
+ title: 'Count'
+ }
+ },
+ type: 'object',
+ required: ['date', 'missed', 'count'],
+ title: 'CalendarDeadlineResponse',
+ description: 'Represents aggregated deadline counts for a specific
calendar time bucket.'
+} as const;
+
export const $CalendarTimeRangeCollectionResponse = {
properties: {
total_entries: {
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index b44b9865237..7f1229ddb9c 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -3,7 +3,7 @@
import type { CancelablePromise } from './core/CancelablePromise';
import { OpenAPI } from './core/OpenAPI';
import { request as __request } from './core/request';
-import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
+import type { GetAssetsData, GetAssetsResponse, GetAssetAliasesData,
GetAssetAliasesResponse, GetAssetAliasData, GetAssetAliasResponse,
GetAssetEventsData, GetAssetEventsResponse, CreateAssetEventData,
CreateAssetEventResponse, MaterializeAssetData, MaterializeAssetResponse,
GetAssetQueuedEventsData, GetAssetQueuedEventsResponse,
DeleteAssetQueuedEventsData, DeleteAssetQueuedEventsResponse, GetAssetData,
GetAssetResponse, GetDagAssetQueuedEventsData, GetDagAssetQueuedEventsResponse,
Dele [...]
export class AssetService {
/**
@@ -5137,6 +5137,39 @@ export class CalendarService {
});
}
+ /**
+ * Get Calendar Deadlines
+ * Get aggregated deadline counts for a Dag, bucketed by deadline_time and
missed status.
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.granularity
+ * @param data.deadlineTimeGte
+ * @param data.deadlineTimeGt
+ * @param data.deadlineTimeLte
+ * @param data.deadlineTimeLt
+ * @returns CalendarDeadlineCollectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static getCalendarDeadlines(data: GetCalendarDeadlinesData):
CancelablePromise<GetCalendarDeadlinesResponse> {
+ return __request(OpenAPI, {
+ method: 'GET',
+ url: '/ui/calendar/{dag_id}/deadlines',
+ path: {
+ dag_id: data.dagId
+ },
+ query: {
+ granularity: data.granularity,
+ deadline_time_gte: data.deadlineTimeGte,
+ deadline_time_gt: data.deadlineTimeGt,
+ deadline_time_lte: data.deadlineTimeLte,
+ deadline_time_lt: data.deadlineTimeLt
+ },
+ errors: {
+ 422: 'Validation Error'
+ }
+ });
+ }
+
}
export class TeamsService {
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index a9d002e1e3f..2d8295b54b9 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -2182,6 +2182,23 @@ export type BaseNodeResponse = {
export type type = 'join' | 'task' | 'asset-condition' | 'asset' |
'asset-alias' | 'asset-name-ref' | 'asset-uri-ref' | 'dag' | 'sensor' |
'trigger';
+/**
+ * Response model for calendar deadline aggregation results.
+ */
+export type CalendarDeadlineCollectionResponse = {
+ total_entries: number;
+ deadlines: Array<CalendarDeadlineResponse>;
+};
+
+/**
+ * Represents aggregated deadline counts for a specific calendar time bucket.
+ */
+export type CalendarDeadlineResponse = {
+ date: string;
+ missed: boolean;
+ count: number;
+};
+
/**
* Response model for calendar time range results.
*/
@@ -4697,6 +4714,17 @@ export type GetCalendarData = {
export type GetCalendarResponse = CalendarTimeRangeCollectionResponse;
+export type GetCalendarDeadlinesData = {
+ dagId: string;
+ deadlineTimeGt?: string | null;
+ deadlineTimeGte?: string | null;
+ deadlineTimeLt?: string | null;
+ deadlineTimeLte?: string | null;
+ granularity?: 'hourly' | 'daily';
+};
+
+export type GetCalendarDeadlinesResponse = CalendarDeadlineCollectionResponse;
+
export type ListTeamsData = {
limit?: number;
offset?: number;
@@ -8551,6 +8579,21 @@ export type $OpenApiTs = {
};
};
};
+ '/ui/calendar/{dag_id}/deadlines': {
+ get: {
+ req: GetCalendarDeadlinesData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: CalendarDeadlineCollectionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
'/ui/teams': {
get: {
req: ListTeamsData;
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py
index 7bd0d69153c..e24ba1814cd 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py
@@ -23,13 +23,17 @@ import pendulum
import pytest
from sqlalchemy.orm import Session
+from airflow._shared.timezones import timezone
+from airflow.models.deadline import Deadline
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import CronPartitionTimetable
+from airflow.sdk.definitions.callback import AsyncCallback
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.state import DagRunState
+from airflow.utils.types import DagRunTriggeredByType, DagRunType
from tests_common.test_utils.asserts import assert_queries_count
-from tests_common.test_utils.db import clear_db_dags, clear_db_runs
+from tests_common.test_utils.db import clear_db_dags, clear_db_deadline,
clear_db_runs
pytestmark = pytest.mark.db_test
@@ -320,3 +324,199 @@ class TestPartitionedCalendar:
assert response.status_code == 200
body = response.json()
assert body == result
+
+
+_CALLBACK_PATH =
"tests.unit.api_fastapi.core_api.routes.ui.test_calendar._noop_callback"
+
+
+async def _noop_callback(**kwargs):
+ """No-op callback used to satisfy Deadline creation in tests."""
+
+
+def _cb() -> AsyncCallback:
+ return AsyncCallback(_CALLBACK_PATH)
+
+
+class TestCalendarDeadlines:
+ """Tests for the GET /calendar/{dag_id}/deadlines endpoint."""
+
+ DAG_NAME = "test_deadlines_calendar_dag"
+
+ @pytest.fixture(autouse=True)
+ def setup_deadlines(self, dag_maker, session) -> None:
+ clear_db_deadline()
+ clear_db_runs()
+ clear_db_dags()
+
+ with dag_maker(self.DAG_NAME, serialized=True, session=session):
+ EmptyOperator(task_id="task")
+
+ run1 = dag_maker.create_dagrun(
+ run_id="run_active",
+ state=DagRunState.SUCCESS,
+ run_type=DagRunType.SCHEDULED,
+ logical_date=timezone.datetime(2025, 1, 1),
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ session.add(
+ Deadline(
+ deadline_time=timezone.datetime(2025, 1, 1, 12, 0, 0),
+ callback=_cb(),
+ dagrun_id=run1.id,
+ deadline_alert_id=None,
+ )
+ )
+
+ run2 = dag_maker.create_dagrun(
+ run_id="run_missed",
+ state=DagRunState.SUCCESS,
+ run_type=DagRunType.SCHEDULED,
+ logical_date=timezone.datetime(2025, 1, 2),
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ missed_dl = Deadline(
+ deadline_time=timezone.datetime(2025, 1, 2, 8, 0, 0),
+ callback=_cb(),
+ dagrun_id=run2.id,
+ deadline_alert_id=None,
+ )
+ missed_dl.missed = True
+ session.add(missed_dl)
+
+ run3 = dag_maker.create_dagrun(
+ run_id="run_multi",
+ state=DagRunState.RUNNING,
+ run_type=DagRunType.SCHEDULED,
+ logical_date=timezone.datetime(2025, 1, 3),
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ for hour in (6, 18):
+ session.add(
+ Deadline(
+ deadline_time=timezone.datetime(2025, 1, 1, hour, 0, 0),
+ callback=_cb(),
+ dagrun_id=run3.id,
+ deadline_alert_id=None,
+ )
+ )
+
+ run4 = dag_maker.create_dagrun(
+ run_id="run_missed_jan1",
+ state=DagRunState.FAILED,
+ run_type=DagRunType.SCHEDULED,
+ logical_date=timezone.datetime(2025, 1, 4),
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ missed_dl_jan1 = Deadline(
+ deadline_time=timezone.datetime(2025, 1, 1, 20, 0, 0),
+ callback=_cb(),
+ dagrun_id=run4.id,
+ deadline_alert_id=None,
+ )
+ missed_dl_jan1.missed = True
+ session.add(missed_dl_jan1)
+
+ dag_maker.sync_dagbag_to_db()
+ session.commit()
+
+ def teardown_method(self) -> None:
+ clear_db_deadline()
+ clear_db_runs()
+ clear_db_dags()
+
+ @pytest.mark.parametrize(
+ ("query_params", "expected"),
+ [
+ (
+ {},
+ {
+ "total_entries": 3,
+ "deadlines": [
+ {"date": "2025-01-01T00:00:00Z", "missed": False,
"count": 3},
+ {"date": "2025-01-01T00:00:00Z", "missed": True,
"count": 1},
+ {"date": "2025-01-02T00:00:00Z", "missed": True,
"count": 1},
+ ],
+ },
+ ),
+ (
+ {
+ "deadline_time_gte": "2025-01-01T00:00:00Z",
+ "deadline_time_lte": "2025-01-01T23:59:59Z",
+ },
+ {
+ "total_entries": 2,
+ "deadlines": [
+ {"date": "2025-01-01T00:00:00Z", "missed": False,
"count": 3},
+ {"date": "2025-01-01T00:00:00Z", "missed": True,
"count": 1},
+ ],
+ },
+ ),
+ (
+ {
+ "deadline_time_gte": "2025-01-02T00:00:00Z",
+ "deadline_time_lte": "2025-01-02T23:59:59Z",
+ },
+ {
+ "total_entries": 1,
+ "deadlines": [
+ {"date": "2025-01-02T00:00:00Z", "missed": True,
"count": 1},
+ ],
+ },
+ ),
+ (
+ {"deadline_time_gte": "2025-06-01T00:00:00Z"},
+ {
+ "total_entries": 0,
+ "deadlines": [],
+ },
+ ),
+ ],
+ )
+ def test_daily_deadlines(self, test_client, query_params, expected):
+ response = test_client.get(f"/calendar/{self.DAG_NAME}/deadlines",
params=query_params)
+ assert response.status_code == 200
+ assert response.json() == expected
+
+ @pytest.mark.parametrize(
+ ("query_params", "expected"),
+ [
+ (
+ {"granularity": "hourly"},
+ {
+ "total_entries": 5,
+ "deadlines": [
+ {"date": "2025-01-01T06:00:00Z", "missed": False,
"count": 1},
+ {"date": "2025-01-01T12:00:00Z", "missed": False,
"count": 1},
+ {"date": "2025-01-01T18:00:00Z", "missed": False,
"count": 1},
+ {"date": "2025-01-01T20:00:00Z", "missed": True,
"count": 1},
+ {"date": "2025-01-02T08:00:00Z", "missed": True,
"count": 1},
+ ],
+ },
+ ),
+ (
+ {
+ "granularity": "hourly",
+ "deadline_time_gte": "2025-01-01T10:00:00Z",
+ "deadline_time_lte": "2025-01-01T20:00:00Z",
+ },
+ {
+ "total_entries": 3,
+ "deadlines": [
+ {"date": "2025-01-01T12:00:00Z", "missed": False,
"count": 1},
+ {"date": "2025-01-01T18:00:00Z", "missed": False,
"count": 1},
+ {"date": "2025-01-01T20:00:00Z", "missed": True,
"count": 1},
+ ],
+ },
+ ),
+ ],
+ )
+ def test_hourly_deadlines(self, test_client, query_params, expected):
+ response = test_client.get(f"/calendar/{self.DAG_NAME}/deadlines",
params=query_params)
+ assert response.status_code == 200
+ assert response.json() == expected
+
+ def test_unknown_dag_returns_empty(self, test_client):
+ response = test_client.get("/calendar/nonexistent_dag/deadlines")
+ assert response.status_code == 200
+ body = response.json()
+ assert body == {"total_entries": 0, "deadlines": []}
