Skip to content

Aip 86: add deadlines to DagRunResponse #50957

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
Open

Aip 86: add deadlines to DagRunResponse #50957

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b9c6156
Add Deadline relationship to DagRun and update Deadline model
rawwar May 21, 2025
980233f
Refactor Deadline relationship in DagRun and Deadline models
rawwar May 21, 2025
e65e392
add test
rawwar May 21, 2025
b07c470
call it deadlines
rawwar May 21, 2025
e9a4c07
remove test
rawwar May 22, 2025
74e28dc
Merge branch 'main' of github.com:apache/airflow into AIP-86/add_dead…
rawwar May 22, 2025
5270ebf
add deadlines to DAGRunResponse
rawwar May 22, 2025
f1d3cf8
update id to be uuid and add tests
rawwar May 22, 2025
a14ec65
fix dag run tests
rawwar May 22, 2025
6b692c3
fix tests
rawwar May 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse
from airflow.api_fastapi.core_api.datamodels.deadline import DeadlineResponse
from airflow.models import DagRun
from airflow.timetables.base import DataInterval
from airflow.utils import timezone
Expand Down Expand Up @@ -68,6 +69,7 @@ class DAGRunResponse(BaseModel):
end_date: datetime | None
data_interval_start: datetime | None
data_interval_end: datetime | None
deadlines: list[DeadlineResponse] | None
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh I missed this. A dag run can have multiple deadlines?

Copy link
Contributor Author

@rawwar rawwar May 22, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, here's an example that @ferruzzi provided:

with DAG(
    dag_id="dag_with_deadlines",
    deadline=[
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=1),
            callback=warn_me_it_may_be_late
            ),
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=2),
            callback=definitely_late
        ),
        DeadlineAlert(
            reference = DeadlineReference.DAGRUN_QUEUED_AT,
            interval=timedelta(hours=3),
            callback=email_my_boss_that_it_wasnt_my_fault
        ),
    ]
)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok and then on the DagResponse we will have the full DeadlineAlert with the reference, interval and callback?

Copy link
Contributor

@bugraoz93 bugraoz93 May 22, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about making this a separate endpoint and calling it for the DagRun retrieved? Simply passing the dag_run_id. This way, we don't increase the DagRunResponse too much. It already has a lot of information. We can have deadline routes as a separate if it is okay from the UI perspective

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think keeping some deadline information on DagRunResponse is valuable because I think "show me recent dag runs with passed deadlines" would be a key feature.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we shouldn't remove rows, maybe just mark them archieved via a boolean in the db. This way we can still access the history for UI or audit. Also adding a column completed_at to know when was the deadline completed can even be an interesting option.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you want to change the plan at this point, please start a dev list discussion for it.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe I misunderstood something. As I understand it, there's no record in the db at all for deadlines unless the run is in progress?

I assume we can always improve later and add that if people request it. (soft delete etc...)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From the API point of view, we can nest the deadline API inside the dagrun one. Because deadline can be considered a sub-resource of the dagrun, they only exist for their corresponding dagrun, and especially when the dagrun is in a specific state, otherwise they are removed. (But that's debatable, since all the API resource are for now flat at root level, we could do the same here).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

deadlines: list[DeadlineResponse] should be enough, the relationship should return an empty list when there is no related items

run_after: datetime
last_scheduling_decision: datetime | None
run_type: DagRunType
Expand Down
32 changes: 32 additions & 0 deletions airflow-core/src/airflow/api_fastapi/core_api/datamodels/deadline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# 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 uuid import UUID

from pydantic import AwareDatetime

from airflow.api_fastapi.core_api.base import BaseModel


class DeadlineResponse(BaseModel):
"""Deadline serializer for responses."""

id: UUID
deadline: AwareDatetime
callback: str | None = None
callback_kwargs: dict | None = None
35 changes: 35 additions & 0 deletions airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -912,6 +912,13 @@ components:
format: date-time
- type: 'null'
title: Data Interval End
deadlines:
anyOf:
- items:
$ref: '#/components/schemas/DeadlineResponse'
type: array
- type: 'null'
title: Deadlines
run_after:
type: string
format: date-time
Expand Down Expand Up @@ -964,6 +971,7 @@ components:
- end_date
- data_interval_start
- data_interval_end
- deadlines
- run_after
- last_scheduling_decision
- run_type
Expand Down Expand Up @@ -1314,6 +1322,33 @@ components:
- queued_dag_count
title: DashboardDagStatsResponse
description: Dashboard DAG Stats serializer for responses.
DeadlineResponse:
properties:
id:
type: string
format: uuid
title: Id
deadline:
type: string
format: date-time
title: Deadline
callback:
anyOf:
- type: string
- type: 'null'
title: Callback
callback_kwargs:
anyOf:
- additionalProperties: true
type: object
- type: 'null'
title: Callback Kwargs
type: object
required:
- id
- deadline
title: DeadlineResponse
description: Deadline serializer for responses.
EdgeResponse:
properties:
source_id:
Expand Down
35 changes: 35 additions & 0 deletions airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8445,6 +8445,13 @@ components:
format: date-time
- type: 'null'
title: Data Interval End
deadlines:
anyOf:
- items:
$ref: '#/components/schemas/DeadlineResponse'
type: array
- type: 'null'
title: Deadlines
run_after:
type: string
format: date-time
Expand Down Expand Up @@ -8497,6 +8504,7 @@ components:
- end_date
- data_interval_start
- data_interval_end
- deadlines
- run_after
- last_scheduling_decision
- run_type
Expand Down Expand Up @@ -8937,6 +8945,33 @@ components:
This is the set of allowable values for the ``warning_type`` field

in the DagWarning model.'
DeadlineResponse:
properties:
id:
type: string
format: uuid
title: Id
deadline:
type: string
format: date-time
title: Deadline
callback:
anyOf:
- type: string
- type: 'null'
title: Callback
callback_kwargs:
anyOf:
- additionalProperties: true
type: object
- type: 'null'
title: Callback Kwargs
type: object
required:
- id
- deadline
title: DeadlineResponse
description: Deadline serializer for responses.
DryRunBackfillCollectionResponse:
properties:
backfills:
Expand Down
57 changes: 57 additions & 0 deletions airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2241,6 +2241,20 @@ export const $DAGRunResponse = {
],
title: "Data Interval End",
},
deadlines: {
anyOf: [
{
items: {
$ref: "#/components/schemas/DeadlineResponse",
},
type: "array",
},
{
type: "null",
},
],
title: "Deadlines",
},
run_after: {
type: "string",
format: "date-time",
Expand Down Expand Up @@ -2330,6 +2344,7 @@ export const $DAGRunResponse = {
"end_date",
"data_interval_start",
"data_interval_end",
"deadlines",
"run_after",
"last_scheduling_decision",
"run_type",
Expand Down Expand Up @@ -2953,6 +2968,48 @@ This is the set of allowable values for the \`\`warning_type\`\` field
in the DagWarning model.`,
} as const;

export const $DeadlineResponse = {
properties: {
id: {
type: "string",
format: "uuid",
title: "Id",
},
deadline: {
type: "string",
format: "date-time",
title: "Deadline",
},
callback: {
anyOf: [
{
type: "string",
},
{
type: "null",
},
],
title: "Callback",
},
callback_kwargs: {
anyOf: [
{
additionalProperties: true,
type: "object",
},
{
type: "null",
},
],
title: "Callback Kwargs",
},
},
type: "object",
required: ["id", "deadline"],
title: "DeadlineResponse",
description: "Deadline serializer for responses.",
} as const;

export const $DryRunBackfillCollectionResponse = {
properties: {
backfills: {
Expand Down
13 changes: 13 additions & 0 deletions airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -575,6 +575,7 @@ export type DAGRunResponse = {
end_date: string | null;
data_interval_start: string | null;
data_interval_end: string | null;
deadlines: Array<DeadlineResponse> | null;
run_after: string;
last_scheduling_decision: string | null;
run_type: DagRunType;
Expand Down Expand Up @@ -764,6 +765,18 @@ export type DagVersionResponse = {
*/
export type DagWarningType = "asset conflict" | "non-existent pool";

/**
* Deadline serializer for responses.
*/
export type DeadlineResponse = {
id: string;
deadline: string;
callback?: string | null;
callback_kwargs?: {
[key: string]: unknown;
} | null;
};

/**
* Backfill collection serializer for responses in dry-run mode.
*/
Expand Down
1 change: 1 addition & 0 deletions airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1196,6 +1196,7 @@ def test_should_respond_200(self, test_client):
"run_after": mock.ANY,
"start_date": None,
"end_date": None,
"deadlines": [],
"data_interval_start": None,
"data_interval_end": None,
"last_scheduling_decision": None,
Expand Down
Loading