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 3dcdcb8002f API: Return asset events for Dag Runs without a start date
(#71379)
3dcdcb8002f is described below
commit 3dcdcb8002fb4babbaa161950d038480f9fcf97e
Author: Ko HanJong <[email protected]>
AuthorDate: Wed Aug 19 03:33:23 2026 +0900
API: Return asset events for Dag Runs without a start date (#71379)
* API: Return asset events for Dag Runs without a start date
Asset events should remain available when their created Dag Run has not
started yet.
* Preserve asset event client coverage
The regression coverage needs to retain started and empty association
responses while exercising unstarted Dag Runs.
* Update generated API metadata
The nullable asset-event response must be represented in the generated
supervisor schema, and CI requires the current dependency metadata in the
lockfile.
* Update TypeScript SDK supervisor schema
Asset events can now reference an unstarted Dag Run, so TypeScript clients
must accept a missing start date.
---
.../api_fastapi/core_api/datamodels/assets.py | 2 +-
.../core_api/openapi/v2-rest-api-generated.yaml | 6 +++--
.../execution_api/datamodels/asset_event.py | 2 +-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 11 ++++++--
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
.../core_api/routes/public/test_assets.py | 26 +++++++++++++++++++
.../core_api/routes/public/test_dag_run.py | 22 +++++++++++++---
.../versions/head/test_asset_events.py | 24 +++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 2 +-
.../tests/airflow_ctl/api/test_operations.py | 22 +++++++++++++---
.../src/airflow/sdk/api/datamodels/_generated.py | 2 +-
.../airflow/sdk/execution_time/schema/schema.json | 13 +++++++---
task-sdk/tests/task_sdk/api/test_client.py | 30 ++++++++++++++++++++--
ts-sdk/src/generated/supervisor.ts | 2 +-
14 files changed, 145 insertions(+), 21 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
index fea14b36755..d4bb4c2ce2b 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
@@ -164,7 +164,7 @@ class DagRunAssetReference(StrictBaseModel):
run_id: str
dag_id: str
logical_date: datetime | None
- start_date: datetime
+ start_date: datetime | None
end_date: datetime | None
state: str
data_interval_start: datetime | None
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 69ac00b6e37..f892c29e770 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -13940,8 +13940,10 @@ components:
- type: 'null'
title: Logical Date
start_date:
- type: string
- format: date-time
+ anyOf:
+ - type: string
+ format: date-time
+ - type: 'null'
title: Start Date
end_date:
anyOf:
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
index f6c3ce82669..7076a50f506 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/asset_event.py
@@ -31,7 +31,7 @@ class DagRunAssetReference(StrictBaseModel):
run_id: str
dag_id: str
logical_date: datetime | None
- start_date: datetime
+ start_date: datetime | None
end_date: datetime | None
state: str
data_interval_start: datetime | None
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 624c749fc6c..d636c35e678 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
@@ -4442,8 +4442,15 @@ export const $DagRunAssetReference = {
title: 'Logical Date'
},
start_date: {
- type: 'string',
- format: 'date-time',
+ anyOf: [
+ {
+ type: 'string',
+ format: 'date-time'
+ },
+ {
+ type: 'null'
+ }
+ ],
title: 'Start Date'
},
end_date: {
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 5afe50bcd14..67b335a8b77 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
@@ -1150,7 +1150,7 @@ export type DagRunAssetReference = {
run_id: string;
dag_id: string;
logical_date: string | null;
- start_date: string;
+ start_date: string | null;
end_date: string | null;
state: string;
data_interval_start: string | null;
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 4179dcb9182..80fa977ee0b 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -1039,6 +1039,32 @@ class TestGetAssetEvents(TestAssets):
assert events[1]["created_dagruns"][0]["triggering"] is False
assert events[2]["created_dagruns"][0]["triggering"] is True
+ def test_should_return_created_dag_run_without_start_date(self,
test_client, session):
+ self.create_assets(num=1, session=session)
+ asset_event = AssetEvent(
+ asset_id=1,
+ source_dag_id="producer_dag",
+ source_run_id="producer_run",
+ timestamp=DEFAULT_DATE,
+ )
+ dag_run = DagRun(
+ dag_id="consumer_dag",
+ run_id="asset-triggered-run",
+ run_type=DagRunType.ASSET_TRIGGERED,
+ logical_date=DEFAULT_DATE,
+ start_date=None,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=DagRunState.QUEUED,
+ )
+ dag_run.consumed_asset_events.append(asset_event)
+ session.add(dag_run)
+ session.commit()
+
+ response = test_client.get("/assets/events")
+
+ assert response.status_code == 200
+ assert
response.json()["asset_events"][0]["created_dagruns"][0]["start_date"] is None
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get("/assets/events")
assert response.status_code == 401
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 4aef320cc63..c2a8528afa6 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -1970,7 +1970,16 @@ class TestGetDagRunAssetTriggerEvents:
["test_partition_key", None],
ids=["partitioned", "non-partitioned"],
)
- def test_should_respond_200(self, partition_key, test_client, dag_maker,
session):
+ @pytest.mark.parametrize(
+ ("state", "start_date_is_none"),
+ [
+ pytest.param(DagRunState.RUNNING, False, id="running"),
+ pytest.param(DagRunState.QUEUED, True,
id="queued-without-start-date"),
+ ],
+ )
+ def test_should_respond_200(
+ self, partition_key, state, start_date_is_none, test_client,
dag_maker, session
+ ):
asset1 = Asset(name="ds1", uri="file:///da1")
# Use PartitionedAtRuntime for partitioned cases so the partition_key
gate does not reject the key.
@@ -2008,6 +2017,9 @@ class TestGetDagRunAssetTriggerEvents:
# explicitly so dag_maker does not try to infer it via
next_dagrun_info (which returns None).
create_dagrun_kwargs["logical_date"] = None
dr = dag_maker.create_dagrun(**create_dagrun_kwargs)
+ dr.state = state
+ if start_date_is_none:
+ dr.start_date = None
dr.consumed_asset_events.append(event)
session.commit()
@@ -2040,8 +2052,12 @@ class TestGetDagRunAssetTriggerEvents:
"data_interval_start":
from_datetime_to_zulu_without_ms(dr.data_interval_start),
"end_date": None,
"logical_date":
from_datetime_to_zulu_without_ms(dr.logical_date),
- "start_date":
from_datetime_to_zulu_without_ms(dr.start_date),
- "state": "running",
+ "start_date": (
+ None
+ if start_date_is_none
+ else
from_datetime_to_zulu_without_ms(dr.start_date)
+ ),
+ "state": state.value,
"partition_key": partition_key,
"triggering": True,
}
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py
index da1e3705daf..8cf59a440c5 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py
@@ -23,6 +23,9 @@ import pytest
from airflow._shared.timezones import timezone
from airflow.models.asset import AssetActive, AssetAliasModel, AssetEvent,
AssetModel
+from airflow.models.dagrun import DagRun
+from airflow.utils.state import DagRunState
+from airflow.utils.types import DagRunType
from tests_common.test_utils.config import conf_vars
@@ -93,6 +96,27 @@ def test_asset_alias(session, test_asset_events, test_asset):
class TestGetAssetEventByAsset:
+ @pytest.mark.usefixtures("test_asset")
+ def test_get_by_asset_with_created_dagrun_without_start_date(self, client,
session, test_asset_events):
+ created_dagrun = DagRun(
+ dag_id="created_dag",
+ run_id="queued_run",
+ logical_date=DEFAULT_DATE,
+ state=DagRunState.QUEUED,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ )
+ test_asset_events[0].created_dagruns.append(created_dagrun)
+ session.commit()
+
+ response = client.get(
+ "/execution/asset-events/by-asset",
+ params={"name": "test_get_asset_by_name", "uri": None},
+ )
+
+ assert response.status_code == 200
+ assert
response.json()["asset_events"][0]["created_dagruns"][0]["start_date"] is None
+
@pytest.mark.parametrize(
("uri", "name"),
[
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 9f464e8bda3..58a5369b1fb 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -527,7 +527,7 @@ class DagRunAssetReference(BaseModel):
run_id: Annotated[str, Field(title="Run Id")]
dag_id: Annotated[str, Field(title="Dag Id")]
logical_date: Annotated[datetime | None, Field(title="Logical Date")]
- start_date: Annotated[datetime, Field(title="Start Date")]
+ start_date: Annotated[datetime | None, Field(title="Start Date")]
end_date: Annotated[datetime | None, Field(title="End Date")]
state: Annotated[str, Field(title="State")]
data_interval_start: Annotated[datetime | None, Field(title="Data Interval
Start")]
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index aaad996ec99..7806fcfb5ab 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -434,14 +434,30 @@ class TestAssetsOperations:
response = client.assets.list_aliases()
assert response == assets_collection_response
- def test_create_event(self):
+ @pytest.mark.parametrize(
+ "created_dagrun",
+ [
+ pytest.param(assets_dag_reference, id="running"),
+ pytest.param(
+ assets_dag_reference.model_copy(
+ update={"start_date": None, "end_date": None, "state":
"queued"}
+ ),
+ id="queued-without-start-date",
+ ),
+ ],
+ )
+ def test_create_event(self, created_dagrun):
+ asset_event_response = self.asset_event_response.model_copy(
+ update={"created_dagruns": [created_dagrun]}
+ )
+
def handle_request(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/api/v2/assets/events"
- return httpx.Response(200,
json=json.loads(self.asset_event_response.model_dump_json()))
+ return httpx.Response(200,
json=json.loads(asset_event_response.model_dump_json()))
client = make_api_client(transport=httpx.MockTransport(handle_request))
response =
client.assets.create_event(asset_event_body=self.asset_create_event_body)
- assert response == self.asset_event_response
+ assert response == asset_event_response
def test_materialize(self):
def handle_request(request: httpx.Request) -> httpx.Response:
diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
index 201f218c3c9..f29a3488440 100644
--- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
+++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
@@ -135,7 +135,7 @@ class DagRunAssetReference(BaseModel):
run_id: Annotated[str, Field(title="Run Id")]
dag_id: Annotated[str, Field(title="Dag Id")]
logical_date: Annotated[AwareDatetime | None, Field(title="Logical Date")]
- start_date: Annotated[AwareDatetime, Field(title="Start Date")]
+ start_date: Annotated[AwareDatetime | None, Field(title="Start Date")]
end_date: Annotated[AwareDatetime | None, Field(title="End Date")]
state: Annotated[str, Field(title="State")]
data_interval_start: Annotated[AwareDatetime | None, Field(title="Data
Interval Start")]
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index 4524c74ff79..85f8b7cfa4e 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -1147,9 +1147,16 @@
"title": "Logical Date"
},
"start_date": {
- "format": "date-time",
- "title": "Start Date",
- "type": "string"
+ "anyOf": [
+ {
+ "format": "date-time",
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "title": "Start Date"
},
"end_date": {
"anyOf": [
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index 8527349f56b..8bd558b461b 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -1202,7 +1202,30 @@ class TestAssetEventOperations:
({"alias_name": "this_asset_alias"}),
],
)
- def test_by_name_get_success(self, request_params):
+ @pytest.mark.parametrize(
+ ("created_dagruns", "expected_created_dagruns"),
+ [
+ pytest.param([], 0, id="without-created-dagrun"),
+ pytest.param(
+ [
+ {
+ "dag_id": "created_dag",
+ "run_id": "queued_run",
+ "logical_date": "2023-01-01T00:00:00Z",
+ "start_date": None,
+ "end_date": None,
+ "state": "queued",
+ "data_interval_start": None,
+ "data_interval_end": None,
+ "partition_key": None,
+ }
+ ],
+ 1,
+ id="queued-created-dagrun-without-start-date",
+ ),
+ ],
+ )
+ def test_by_name_get_success(self, request_params, created_dagruns,
expected_created_dagruns):
def handle_request(request: httpx.Request) -> httpx.Response:
params = request.url.params
if request.url.path == "/asset-events/by-asset":
@@ -1224,7 +1247,7 @@ class TestAssetEventOperations:
"uri": "s3://bucket/key",
"group": "asset",
},
- "created_dagruns": [],
+ "created_dagruns": created_dagruns,
"timestamp": "2023-01-01T00:00:00Z",
}
]
@@ -1238,6 +1261,9 @@ class TestAssetEventOperations:
assert len(result.asset_events) == 1
assert result.asset_events[0].asset.name == "this_asset"
assert result.asset_events[0].asset.uri == "s3://bucket/key"
+ assert len(result.asset_events[0].created_dagruns) ==
expected_created_dagruns
+ if expected_created_dagruns:
+ assert result.asset_events[0].created_dagruns[0].start_date is None
def test_partition_key_exact_match_param_passed(self):
def handle_request(request: httpx.Request) -> httpx.Response:
diff --git a/ts-sdk/src/generated/supervisor.ts
b/ts-sdk/src/generated/supervisor.ts
index e12e5d815ae..8eb27a1d815 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -37,7 +37,7 @@ export type Extra1 = {
export type RunId = string;
export type DagId = string;
export type LogicalDate = string | null;
-export type StartDate = string;
+export type StartDate = string | null;
export type EndDate = string | null;
export type State = string;
export type DataIntervalStart = string | null;
