This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 1ce3f17906c Add the option to select bundle version parameter on dag
run trigger endpoint (#61550)
1ce3f17906c is described below
commit 1ce3f17906c1d2356abeb21d319ad144f016c8bd
Author: Itay Adler <[email protected]>
AuthorDate: Fri Jul 10 19:54:34 2026 +0300
Add the option to select bundle version parameter on dag run trigger
endpoint (#61550)
* feat: adding the option to add bundle version parameter on dag run
trigger endpoint
* removing code comments and adding the new parameter to the schema
* feat: adding the new parameter to schema and types files
* fix: fixing format and missing field in class
* fix: fixing the usage of bundle version in orm dagrun creation
* feat: added migration file
* feat: adding unit test for bundle version parameter dagrun endpoint
* fix: fixing the unit test by removing unecessary asserts
* fix: removing duplicate dag version check and raising exception
accordingly
* fix: adding a raise exception when passing incorrect bundle version
* fix: verify exception message to raise accordingly
* fix: wiring to bundle version and moved to the correct version target
* fix: missed adding bundle_version field in 2 unit tests response assertion
* fix: adding missing bundle version field in tests
* fix: handling dag and dag's tasks correctly when passing bundle version
* feat: added new airflow exception to handle no dag version has found
* fix: assigning run dag to dag version when only using bundle version
* fix: fixing assertion equal failing
* fix: removed unecessary init files after prek hook added them
* fix: adding support for all the dag-level callbacks
* feat: adding versioning test for bundle version parameter
* fix: rebasing branch
* feat: adding the option to add bundle version parameter on dag run
trigger endpoint
* removing code comments and adding the new parameter to the schema
* feat: adding the new parameter to schema and types files
* fix: fixing format and missing field in class
* fix: fixing the usage of bundle version in orm dagrun creation
* feat: added migration file
* feat: adding unit test for bundle version parameter dagrun endpoint
* fix: fixing the unit test by removing unecessary asserts
* fix: removing duplicate dag version check and raising exception
accordingly
* fix: adding a raise exception when passing incorrect bundle version
* fix: verify exception message to raise accordingly
* fix: wiring to bundle version and moved to the correct version target
* fix: missed adding bundle_version field in 2 unit tests response assertion
* fix: adding missing bundle version field in tests
* fix: handling dag and dag's tasks correctly when passing bundle version
* feat: added new airflow exception to handle no dag version has found
* fix: assigning run dag to dag version when only using bundle version
* fix: fixing assertion equal failing
* fix: removed unecessary init files after prek hook added them
* fix: adding support for all the dag-level callbacks
* fix: prek didn't updated datamodels for airflowctl
* fix: adding bundle version support also in asset route, fixing execution
api versioning, and also getting latest version of dag with bundle version
* fix: duplicate import and duplicate methods
* fix: remove duplicate DagVersion lookup and unintended dag swap for
non-explicit triggers
* fix: reverting uv lock due to unnecessary changes
* fix: fixing test to resolve ci errors
* fix: restore uv.lock to upstream/main
* fix: remove double code, moving to new CalVer file, raising correct
exception and add more callbacks support in test
* fix: instead of always using latest dag using the correct dag if bundle
version is passed
* fix: missing fields in schema.json
* Regenerate supervisor schema snapshot after rebase onto main
The schema.json was stale: it contained legacy top-level $defs
(DagRun, DagRunState, DagRunType, JsonValue, and nested asset
reference types) that the generator no longer emits after upstream
main restructured them. bundle_version also needed to land at the
correct offset in the generated output.
* fix: consolidate bundle version resolution, fix double deserialization,
drop bundle_version from Execution API
* restored provider_dependencies.json.sha256sum
* Fix bundle_version overwritten by sync_dag_to_db and stale supervisor
schema snapshot
* revert accidently changes to uv.lock
* Move AddPartitionDateField to 2026-06-30 version to match partition_date
introduction timeline
* Remove dead AirflowBadRequest handler and unintended execution API
version split
* Derive run context from requested bundle version on trigger
* Validate partition_key and allowed_run_types against the requested bundle
version
* Revert accidental uv.lock
* Regenerate private UI OpenAPI spec
* Removing unnecessary callback attributes
* Remove unnecessary test due to remove bundle version from execution api
---------
Co-authored-by: Itay Adler <[email protected]>
Co-authored-by: Itay Adler <[email protected]>
---
.../api_fastapi/core_api/datamodels/dag_run.py | 1 +
.../core_api/openapi/v2-rest-api-generated.yaml | 10 +
.../api_fastapi/core_api/routes/public/assets.py | 36 +++-
.../api_fastapi/core_api/routes/public/dag_run.py | 58 ++++--
airflow-core/src/airflow/exceptions.py | 4 +
airflow-core/src/airflow/models/dag_version.py | 9 +-
.../src/airflow/serialization/definitions/dag.py | 57 ++++--
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 22 +++
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +
.../core_api/routes/public/test_assets.py | 88 +++++++++
.../core_api/routes/public/test_dag_run.py | 217 ++++++++++++++++++++-
airflow-core/tests/unit/models/test_dag.py | 118 +++++++++++
.../src/airflowctl/api/datamodels/generated.py | 2 +
devel-common/src/tests_common/test_utils/dag.py | 10 +-
14 files changed, 593 insertions(+), 41 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
index 91a60cb776f..cae9d3a8339 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
@@ -228,6 +228,7 @@ class TriggerDAGRunPostBody(StrictBaseModel):
conf: dict | None = Field(default_factory=dict)
note: str | None = None
partition_key: str | None = None
+ bundle_version: str | None = None
@model_validator(mode="after")
def check_data_intervals(self):
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 d3a1d124943..005da0c6f4b 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
@@ -15274,6 +15274,11 @@ components:
- type: string
- type: 'null'
title: Partition Key
+ bundle_version:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Bundle Version
additionalProperties: false
type: object
title: MaterializeAssetBody
@@ -16807,6 +16812,11 @@ components:
- type: string
- type: 'null'
title: Partition Key
+ bundle_version:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Bundle Version
additionalProperties: false
type: object
required:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index d83ed8ec8d8..424938af86d 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -75,7 +75,7 @@ from airflow.api_fastapi.core_api.security import (
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.assets.manager import asset_manager
from airflow.configuration import conf
-from airflow.exceptions import ParamValidationError
+from airflow.exceptions import DagVersionNotFound, ParamValidationError
from airflow.models.asset import (
AssetAliasModel,
AssetDagRunQueue,
@@ -84,6 +84,7 @@ from airflow.models.asset import (
AssetWatcherModel,
TaskOutletAssetReference,
)
+from airflow.models.dag_version import DagVersion
from airflow.typing_compat import Unpack
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -458,14 +459,31 @@ def materialize_asset(
dag = get_latest_version_of_dag(dag_bag, dag_id, session)
- if dag.allowed_run_types is not None and DagRunType.ASSET_MATERIALIZATION
not in dag.allowed_run_types:
- raise HTTPException(
- status.HTTP_400_BAD_REQUEST,
- f"Dag with dag_id: '{dag_id}' does not allow asset materialization
runs",
- )
+ resolved_body = body or MaterializeAssetBody()
try:
- params = (body or MaterializeAssetBody()).validate_context(dag)
+ preloaded_dag_version = None
+ context_dag = dag
+ if resolved_body.bundle_version is not None and not
dag.disable_bundle_versioning:
+ preloaded_dag_version = DagVersion.get_latest_version(
+ dag_id, bundle_version=resolved_body.bundle_version,
load_serialized_dag=True, session=session
+ )
+ if not preloaded_dag_version:
+ raise DagVersionNotFound(
+ f"DAG with dag_id: '{dag_id}' does not have a version for
bundle_version '{resolved_body.bundle_version}'"
+ )
+ context_dag = preloaded_dag_version.serialized_dag.dag
+
+ if (
+ context_dag.allowed_run_types is not None
+ and DagRunType.ASSET_MATERIALIZATION not in
context_dag.allowed_run_types
+ ):
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Dag with dag_id: '{dag_id}' does not allow asset
materialization runs",
+ )
+
+ params = resolved_body.validate_context(context_dag)
return dag.create_dagrun(
run_id=params["run_id"],
logical_date=params["logical_date"],
@@ -480,9 +498,13 @@ def materialize_asset(
partition_date=params["partition_date"],
note=params["note"],
session=session,
+ bundle_version=resolved_body.bundle_version,
+ dag_version=preloaded_dag_version,
)
except (ParamValidationError, ValueError) as e:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
+ except DagVersionNotFound as e:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
@assets_router.get(
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index 278f6e99119..59b3593b9ed 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -108,7 +108,7 @@ from airflow.api_fastapi.core_api.services.public.dag_run
import (
perform_clear_dag_run,
)
from airflow.api_fastapi.logging.decorators import action_logging
-from airflow.exceptions import ParamValidationError
+from airflow.exceptions import DagVersionNotFound, ParamValidationError
from airflow.models import DagModel, DagRun
from airflow.models.asset import AssetEvent
from airflow.models.dag_version import DagVersion
@@ -693,21 +693,44 @@ def trigger_dag_run(
f"Dag with dag_id: '{dag_id}' has import errors and cannot be
triggered",
)
- if dm.allowed_run_types is not None and DagRunType.MANUAL not in
dm.allowed_run_types:
- raise HTTPException(
- status.HTTP_400_BAD_REQUEST,
- f"Dag with dag_id: '{dag_id}' does not allow manual runs",
- )
-
referer = request.headers.get("referer")
if referer:
triggered_by = DagRunTriggeredByType.UI
else:
triggered_by = DagRunTriggeredByType.REST_API
- dag = get_latest_version_of_dag(dag_bag, dag_id, session)
try:
- params = body.validate_context(dag)
+ dag = get_latest_version_of_dag(dag_bag, dag_id, session)
+ preloaded_dag_version = None
+ context_dag = dag
+ if body.bundle_version is not None and not
dag.disable_bundle_versioning:
+ preloaded_dag_version = DagVersion.get_latest_version(
+ dag_id, bundle_version=body.bundle_version,
load_serialized_dag=True, session=session
+ )
+ if not preloaded_dag_version:
+ raise DagVersionNotFound(
+ f"DAG with dag_id: '{dag_id}' does not have a version for
bundle_version '{body.bundle_version}'"
+ )
+ context_dag = preloaded_dag_version.serialized_dag.dag
+
+ if (
+ context_dag.allowed_run_types is not None
+ and DagRunType.MANUAL not in context_dag.allowed_run_types
+ ):
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Dag with dag_id: '{dag_id}' does not allow manual runs",
+ )
+
+ params = body.validate_context(context_dag)
+
+ if body.bundle_version is not None:
+ if dag.disable_bundle_versioning:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"DAG with dag_id: '{dag_id}' does not support bundle
versioning",
+ )
+
dag_run = dag.create_dagrun(
run_id=params["run_id"],
logical_date=params["logical_date"],
@@ -719,17 +742,22 @@ def trigger_dag_run(
triggering_user_name=user.get_name(),
state=DagRunState.QUEUED,
partition_key=params["partition_key"],
+ bundle_version=body.bundle_version,
+ dag_version=preloaded_dag_version,
partition_date=params["partition_date"],
session=session,
)
+
+ dag_run_note = body.note
+ if dag_run_note:
+ current_user_id = user.get_id()
+ dag_run.note = (dag_run_note, current_user_id)
+ return dag_run
+
except (ParamValidationError, ValueError) as e:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e
-
- dag_run_note = body.note
- if dag_run_note:
- current_user_id = user.get_id()
- dag_run.note = (dag_run_note, current_user_id)
- return dag_run
+ except DagVersionNotFound as e:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
@dag_run_router.get(
diff --git a/airflow-core/src/airflow/exceptions.py
b/airflow-core/src/airflow/exceptions.py
index addd4f8f2a5..cb45b050fce 100644
--- a/airflow-core/src/airflow/exceptions.py
+++ b/airflow-core/src/airflow/exceptions.py
@@ -141,6 +141,10 @@ class DagRunNotFound(AirflowNotFoundException):
"""Raise when a DAG Run is not available in the system."""
+class DagVersionNotFound(AirflowNotFoundException):
+ """Raised when a DagVersion for the given dag_id / bundle_version is not
found."""
+
+
class DagNotPartitionedError(ValueError):
"""Raise when a partition_key is supplied for a Dag that is not
partitioned."""
diff --git a/airflow-core/src/airflow/models/dag_version.py
b/airflow-core/src/airflow/models/dag_version.py
index eb0a5204956..8aa7efcab01 100644
--- a/airflow-core/src/airflow/models/dag_version.py
+++ b/airflow-core/src/airflow/models/dag_version.py
@@ -154,6 +154,7 @@ class DagVersion(Base):
bundle_version: str | None = None,
load_dag_model: bool = False,
load_bundle_model: bool = False,
+ load_serialized_dag: bool = False,
) -> Select:
"""
Get the select object to get the latest version of the DAG.
@@ -162,7 +163,7 @@ class DagVersion(Base):
:return: The select object.
"""
query = select(cls).where(cls.dag_id == dag_id)
- if bundle_version:
+ if bundle_version is not None:
query = query.where(cls.bundle_version == bundle_version)
if load_dag_model:
@@ -171,6 +172,9 @@ class DagVersion(Base):
if load_bundle_model:
query = query.options(joinedload(cls.bundle))
+ if load_serialized_dag:
+ query = query.options(joinedload(cls.serialized_dag))
+
# Order by version_number, not created_at: version_number is monotonic
and unique per
# dag_id, so it is deterministic even when two versions share a
created_at timestamp.
# write_dag relies on this select to compute the next version_number;
ordering by
@@ -188,6 +192,7 @@ class DagVersion(Base):
bundle_version: str | None = None,
load_dag_model: bool = False,
load_bundle_model: bool = False,
+ load_serialized_dag: bool = False,
session: Session = NEW_SESSION,
) -> DagVersion | None:
"""
@@ -197,6 +202,7 @@ class DagVersion(Base):
:param session: The database session.
:param load_dag_model: Whether to load the DAG model.
:param load_bundle_model: Whether to load the DagBundle model.
+ :param load_serialized_dag: Whether to eagerly load the serialized DAG.
:return: The latest version of the DAG or None if not found.
"""
return session.scalar(
@@ -205,6 +211,7 @@ class DagVersion(Base):
bundle_version=bundle_version,
load_dag_model=load_dag_model,
load_bundle_model=load_bundle_model,
+ load_serialized_dag=load_serialized_dag,
)
)
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py
b/airflow-core/src/airflow/serialization/definitions/dag.py
index abae0f97785..8ed0fee2cca 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -36,6 +36,7 @@ from airflow.configuration import conf as airflow_conf
from airflow.exceptions import (
AirflowException,
DagNotPartitionedError,
+ DagVersionNotFound,
InvalidPartitionKeyError,
NodeNotFound,
TaskNotFound,
@@ -583,8 +584,10 @@ class SerializedDAG:
creating_job_id: int | None = None,
backfill_id: NonNegativeInt | None = None,
partition_key: str | None = None,
+ bundle_version: str | None = None,
partition_date: datetime.datetime | None = None,
note: str | None = None,
+ dag_version: DagVersion | None = None,
session: Session = NEW_SESSION,
) -> DagRun:
"""
@@ -652,10 +655,28 @@ class SerializedDAG:
f"is reserved for {inferred_run_type.value} runs"
)
- self.validate_partition_key(partition_key)
-
# todo: AIP-78 add verification that if run type is backfill then we
have a backfill id
- copied_params = self.params.deep_merge(conf)
+
+ # When triggering against a specific bundle version, resolve that
version first so
+ # partition_key and conf are validated against it (not the live/latest
dag).
+ if bundle_version is not None:
+ if self.disable_bundle_versioning:
+ raise ValueError(f"DAG with dag_id: '{self.dag_id}' does not
support bundle versioning")
+ if dag_version is None:
+ dag_version = DagVersion.get_latest_version(
+ self.dag_id, bundle_version=bundle_version,
load_serialized_dag=True, session=session
+ )
+ if not dag_version:
+ raise DagVersionNotFound(
+ f"DAG with dag_id: '{self.dag_id}' does not have a
version for bundle_version '{bundle_version}'"
+ )
+ params_dag = dag_version.serialized_dag.dag
+ else:
+ params_dag = self
+
+ params_dag.validate_partition_key(partition_key)
+
+ copied_params = params_dag.params.deep_merge(conf)
copied_params.validate()
orm_dagrun = _create_orm_dagrun(
dag=self,
@@ -672,12 +693,15 @@ class SerializedDAG:
triggered_by=triggered_by,
triggering_user_name=triggering_user_name,
partition_key=partition_key,
+ bundle_version=bundle_version,
partition_date=partition_date,
note=note,
+ dag_version=dag_version,
+ resolved_dag=params_dag if bundle_version is not None else None,
session=session,
)
- if self.deadline:
+ if params_dag.deadline:
self._process_dagrun_deadline_alerts(orm_dagrun, session)
return orm_dagrun
@@ -1393,16 +1417,25 @@ def _create_orm_dagrun(
triggered_by: DagRunTriggeredByType,
triggering_user_name: str | None = None,
partition_key: str | None = None,
+ bundle_version: str | None = None,
partition_date: datetime.datetime | None = None,
note: str | None = None,
+ dag_version: DagVersion | None = None,
+ resolved_dag: SerializedDAG | None = None,
session: Session = NEW_SESSION,
) -> DagRun:
- bundle_version = None
- if not dag.disable_bundle_versioning:
- bundle_version = session.scalar(
- select(DagModel.bundle_version).where(DagModel.dag_id ==
dag.dag_id),
- )
- dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
+ resolved_bundle_version: str | None = None
+ use_resolved_dag = False
+ if dag_version is not None:
+ resolved_bundle_version = bundle_version
+ use_resolved_dag = True
+ else:
+ if not dag.disable_bundle_versioning:
+ resolved_bundle_version = session.scalar(
+ select(DagModel.bundle_version).where(DagModel.dag_id ==
dag.dag_id)
+ )
+ dag_version = DagVersion.get_latest_version(dag.dag_id,
session=session)
+
if not dag_version:
raise AirflowException(f"Cannot create DagRun for DAG {dag.dag_id}
because the dag is not serialized")
@@ -1420,7 +1453,7 @@ def _create_orm_dagrun(
triggered_by=triggered_by,
triggering_user_name=triggering_user_name,
backfill_id=backfill_id,
- bundle_version=bundle_version,
+ bundle_version=resolved_bundle_version,
partition_key=partition_key,
partition_date=partition_date,
note=note,
@@ -1433,6 +1466,8 @@ def _create_orm_dagrun(
session.add(run)
session.flush()
run.dag = dag
+ if use_resolved_dag:
+ run.dag = resolved_dag if resolved_dag is not None else
dag_version.serialized_dag.dag
# create the associated task instances
# state is None at the moment of creation
run.verify_integrity(session=session, dag_version_id=dag_version.id)
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 df738113acc..e1880fdd4ea 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
@@ -5652,6 +5652,17 @@ export const $MaterializeAssetBody = {
}
],
title: 'Partition Key'
+ },
+ bundle_version: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Bundle Version'
}
},
additionalProperties: false,
@@ -7959,6 +7970,17 @@ export const $TriggerDAGRunPostBody = {
}
],
title: 'Partition Key'
+ },
+ bundle_version: {
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
+ title: 'Bundle Version'
}
},
additionalProperties: false,
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 111ba9204ef..750202cd7fb 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
@@ -1482,6 +1482,7 @@ export type MaterializeAssetBody = {
} | null;
note?: string | null;
partition_key?: string | null;
+ bundle_version?: string | null;
};
/**
@@ -1989,6 +1990,7 @@ export type TriggerDAGRunPostBody = {
} | null;
note?: string | null;
partition_key?: string | null;
+ bundle_version?: 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 1ffa8552b84..d5c56d133ad 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
@@ -1763,6 +1763,43 @@ class TestPostAssetMaterialize(TestAssets):
== f"Dag with dag_id: '{self.DAG_ASSET1_ID}' does not allow asset
materialization runs"
)
+ def test_materialize_allowed_run_types_from_requested_version(self,
test_client, session, dag_maker):
+ """Asset materialization allowed_run_types is enforced from the
requested bundle version, not latest."""
+ bundle_name = "allowed_run_types_bundle"
+ asset = session.get(AssetModel, 1).to_serialized()
+
+ with dag_maker(
+ self.DAG_ASSET1_ID,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ schedule=None,
+ session=session,
+ ):
+ EmptyOperator(task_id="task_v1", outlets=asset)
+
+ with dag_maker(
+ self.DAG_ASSET1_ID,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ schedule="@daily",
+ allowed_run_types=[DagRunType.SCHEDULED],
+ session=session,
+ ):
+ EmptyOperator(task_id="task_v2", outlets=asset)
+
+ # v1 allows materialization; latest v2 does not. Requesting v1 must
succeed.
+ response = test_client.post("/assets/1/materialize",
json={"bundle_version": "v1"})
+ assert response.status_code == 200
+ assert response.json()["bundle_version"] == "v1"
+
+ # Without bundle_version the latest (v2) governs and rejects the run.
+ response = test_client.post("/assets/1/materialize")
+ assert response.status_code == 400
+ assert (
+ response.json()["detail"]
+ == f"Dag with dag_id: '{self.DAG_ASSET1_ID}' does not allow asset
materialization runs"
+ )
+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
def test_should_respond_403_when_user_cannot_trigger_dag(self,
test_client):
with mock.patch(
@@ -1784,6 +1821,57 @@ class TestPostAssetMaterialize(TestAssets):
user=mock.ANY,
)
+ def test_should_respond_with_bundle_version(self, test_client, session,
dag_maker):
+ """Test that asset materialization respects bundle_version
parameter."""
+ bundle_name = "testing_bundle"
+ asset = session.get(AssetModel, 1).to_serialized()
+
+ with dag_maker(
+ self.DAG_ASSET1_ID,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ schedule=None,
+ session=session,
+ ):
+ EmptyOperator(task_id="task_v1", outlets=asset)
+
+ with dag_maker(
+ self.DAG_ASSET1_ID,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ schedule=None,
+ session=session,
+ ):
+ EmptyOperator(task_id="task_v2", outlets=asset)
+
+ response = test_client.post("/assets/1/materialize",
json={"bundle_version": "v1"})
+ assert response.status_code == 200
+ assert response.json()["bundle_version"] == "v1"
+
+ response = test_client.post("/assets/1/materialize",
json={"bundle_version": "invalid_version"})
+ assert response.status_code == 404
+ assert (
+ f"DAG with dag_id: '{self.DAG_ASSET1_ID}' does not have a version
for bundle_version 'invalid_version'"
+ in response.json()["detail"]
+ )
+
+ with dag_maker(
+ self.DAG_ASSET1_ID,
+ bundle_name=bundle_name,
+ bundle_version="v3",
+ schedule=None,
+ session=session,
+ ):
+ EmptyOperator(task_id="task_v3", outlets=asset)
+ dag_maker.dag.disable_bundle_versioning = True
+
+ response = test_client.post("/assets/1/materialize",
json={"bundle_version": "v1"})
+ assert response.status_code == 400
+ assert (
+ f"DAG with dag_id: '{self.DAG_ASSET1_ID}' does not support bundle
versioning"
+ in response.json()["detail"]
+ )
+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
def test_should_respond_400_on_invalid_dag_run_id(self, test_client):
"""A dag_run_id containing '..' triggers ValueError in
DagRun.validate_run_id.
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 d21219156aa..3b1422c79c6 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
@@ -3293,12 +3293,22 @@ class TestTriggerDagRun:
== "Dag with dag_id: 'import_errors' has import errors and cannot
be triggered"
)
- def test_should_respond_400_if_manual_runs_denied(self, test_client,
session, testing_dag_bundle):
+ def test_should_respond_400_if_manual_runs_denied(self, test_client,
session, dag_maker):
now = timezone.utcnow().isoformat()
- self._dags_for_trigger_tests(session)
- response = test_client.post("/dags/allowed_scheduled/dagRuns",
json={"logical_date": now})
+ dag_id = "allowed_scheduled"
+ with dag_maker(
+ dag_id=dag_id,
+ schedule="@daily",
+ allowed_run_types=[DagRunType.SCHEDULED],
+ session=session,
+ serialized=True,
+ ):
+ EmptyOperator(task_id="task")
+ session.commit()
+
+ response = test_client.post(f"/dags/{dag_id}/dagRuns",
json={"logical_date": now})
assert response.status_code == 400
- assert response.json()["detail"] == "Dag with dag_id:
'allowed_scheduled' does not allow manual runs"
+ assert response.json()["detail"] == f"Dag with dag_id: '{dag_id}' does
not allow manual runs"
@time_machine.travel(timezone.utcnow(), tick=False)
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
@@ -3505,6 +3515,205 @@ class TestTriggerDagRun:
run = session.scalars(select(DagRun).where(DagRun.run_id ==
run_id_without_logical_date)).one()
assert run.dag_id == custom_dag_id
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_trigger_dag_run_with_bundle_version(self, test_client, session,
dag_maker):
+ """Test triggering a DAG run with a specific bundle version."""
+ from tests_common.test_utils.dag import sync_dag_to_db
+
+ dag_id = "test_bundle_version_dag"
+ bundle_name = "testing_bundle"
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ session=session,
+ ) as dag1:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag1, bundle_name=bundle_name, bundle_version="v1")
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ session=session,
+ ) as dag2:
+ EmptyOperator(task_id="task_1")
+ EmptyOperator(task_id="task_2")
+ sync_dag_to_db(dag2, bundle_name=bundle_name, bundle_version="v2")
+
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns", json={"logical_date":
"2024-01-01T00:00:00Z", "bundle_version": "v1"}
+ )
+ assert response.status_code == 200
+ assert response.json()["dag_versions"][0]["bundle_version"] == "v1"
+ run_id_v1 = response.json()["dag_run_id"]
+ dr_v1 = session.scalars(select(DagRun).where(DagRun.run_id ==
run_id_v1)).one()
+ assert {ti.task_id for ti in dr_v1.task_instances} == {"task_1"}
+
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={
+ "logical_date": "2024-01-02T00:00:00Z",
+ },
+ )
+ assert response.status_code == 200
+ assert response.json()["dag_versions"][0]["bundle_version"] == "v2"
+ run_id_v2 = response.json()["dag_run_id"]
+ dr_v2 = session.scalars(select(DagRun).where(DagRun.run_id ==
run_id_v2)).one()
+ assert {ti.task_id for ti in dr_v2.task_instances} == {"task_1",
"task_2"}
+
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={"logical_date": "2024-01-03T00:00:00Z", "bundle_version":
"invalid_version"},
+ )
+ assert response.status_code == 404
+ assert (
+ f"DAG with dag_id: '{dag_id}' does not have a version for
bundle_version 'invalid_version'"
+ in response.json()["detail"]
+ )
+
+ dag2.disable_bundle_versioning = True
+ sync_dag_to_db(dag2, bundle_name=bundle_name)
+
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns", json={"logical_date":
"2024-01-04T00:00:00Z", "bundle_version": "v1"}
+ )
+ assert response.status_code == 400
+ assert f"DAG with dag_id: '{dag_id}' does not support bundle
versioning" in response.json()["detail"]
+
+ def test_trigger_dag_run_bundle_version_validates_against_old_param_schema(
+ self, test_client, session, dag_maker
+ ):
+ """Conf is validated against the requested bundle version's param
schema, not the live dag's."""
+ from tests_common.test_utils.dag import sync_dag_to_db
+
+ dag_id = "test_bundle_param_schema_dag"
+ bundle_name = "param_schema_bundle"
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ session=session,
+ params={"env": Param("staging", type="string", enum=["staging",
"prod"])},
+ ) as dag1:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag1, bundle_name=bundle_name, bundle_version="v1")
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ session=session,
+ params={"env": Param("dev", type="string", enum=["dev", "staging",
"prod"])},
+ ) as dag2:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag2, bundle_name=bundle_name, bundle_version="v2")
+
+ # "dev" is valid for v2 but not for v1's enum — triggering v1 should
reject it.
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={"logical_date": "2024-02-01T00:00:00Z", "bundle_version":
"v1", "conf": {"env": "dev"}},
+ )
+ assert response.status_code == 400
+
+ # "staging" is valid for both v1 and v2 — triggering v1 should accept
it.
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={
+ "logical_date": "2024-02-02T00:00:00Z",
+ "bundle_version": "v1",
+ "conf": {"env": "staging"},
+ },
+ )
+ assert response.status_code == 200
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_trigger_dag_run_bundle_version_uses_v1_timetable(self,
test_client, session, dag_maker):
+ """Triggering with bundle_version='v1' must derive data_interval from
v1's timetable, not v2's."""
+ from tests_common.test_utils.dag import sync_dag_to_db
+
+ dag_id = "test_bundle_timetable_dag"
+ bundle_name = "timetable_bundle"
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ schedule=CronDataIntervalTimetable("0 0 * * *", timezone="UTC"),
+ session=session,
+ ) as dag1:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag1, bundle_name=bundle_name, bundle_version="v1")
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ schedule=None,
+ session=session,
+ ) as dag2:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag2, bundle_name=bundle_name, bundle_version="v2")
+
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={"logical_date": "2024-01-01T00:00:00Z", "bundle_version":
"v1"},
+ )
+ assert response.status_code == 200
+ data = response.json()
+ assert data["dag_versions"][0]["bundle_version"] == "v1"
+ # data_interval must come from v1's daily cron timetable, not v2's
null timetable.
+ # For a "0 0 * * *" cron, logical_date is the interval END, so
interval is [prev_day, logical_date].
+ assert data["data_interval_start"] == "2023-12-31T00:00:00Z"
+ assert data["data_interval_end"] == "2024-01-01T00:00:00Z"
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_trigger_dag_run_allowed_run_types_from_requested_version(self,
test_client, session, dag_maker):
+ """allowed_run_types is enforced from the requested bundle version,
not the latest."""
+ from tests_common.test_utils.dag import sync_dag_to_db
+
+ dag_id = "test_bundle_allowed_run_types_dag"
+ bundle_name = "allowed_run_types_bundle"
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v1",
+ schedule="@daily",
+ allowed_run_types=[DagRunType.MANUAL, DagRunType.SCHEDULED],
+ session=session,
+ ) as dag1:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag1, bundle_name=bundle_name, bundle_version="v1")
+
+ with dag_maker(
+ dag_id=dag_id,
+ bundle_name=bundle_name,
+ bundle_version="v2",
+ schedule="@daily",
+ allowed_run_types=[DagRunType.SCHEDULED],
+ session=session,
+ ) as dag2:
+ EmptyOperator(task_id="task_1")
+ sync_dag_to_db(dag2, bundle_name=bundle_name, bundle_version="v2")
+
+ # Latest (v2) disallows manual runs; v1 allows them. Triggering v1
must succeed.
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={"logical_date": "2024-02-01T00:00:00Z", "bundle_version":
"v1"},
+ )
+ assert response.status_code == 200
+
+ # Without bundle_version the latest (v2) governs and rejects the
manual run.
+ response = test_client.post(
+ f"/dags/{dag_id}/dagRuns",
+ json={"logical_date": "2024-02-02T00:00:00Z"},
+ )
+ assert response.status_code == 400
+ assert response.json()["detail"] == f"Dag with dag_id: '{dag_id}' does
not allow manual runs"
+
def
test_should_respond_400_when_partition_key_given_for_non_partitioned_dag(self,
test_client):
"""Passing partition_key to a non-partitioned Dag via REST trigger
must return 400, not 500.
diff --git a/airflow-core/tests/unit/models/test_dag.py
b/airflow-core/tests/unit/models/test_dag.py
index 81198d8e418..6a1e9e5782c 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -1581,6 +1581,38 @@ class TestDag:
partition_key=123,
)
+ def
test_create_dagrun_partition_key_validated_against_requested_version(self,
dag_maker, session):
+ """create_dagrun validates partition_key against the requested bundle
version, not the latest."""
+ dag_id = "test_create_dagrun_partition_key_bundle_version"
+
+ with dag_maker(
+ dag_id,
+ schedule=CronPartitionTimetable("@daily", timezone="UTC"),
+ bundle_version="v1",
+ session=session,
+ ):
+ EmptyOperator(task_id="task")
+
+ with dag_maker(dag_id, schedule=None, bundle_version="v2",
session=session):
+ EmptyOperator(task_id="task")
+ session.commit()
+
+ scheduler_dag_v2 = dag_maker.serialized_dag
+
+ # Latest (v2) is not partitioned, but the requested v1 is: the key
must be
+ # accepted against v1 rather than rejected against the latest dag.
+ dr = scheduler_dag_v2.create_dagrun(
+ run_id="manual__partition_key_from_v1",
+ run_after=DEFAULT_DATE,
+ run_type=DagRunType.MANUAL,
+ state=State.NONE,
+ triggered_by=DagRunTriggeredByType.TEST,
+ partition_key="my-key",
+ bundle_version="v1",
+ session=session,
+ )
+ assert dr.partition_key == "my-key"
+
@pytest.mark.need_serialized_dag
@pytest.mark.parametrize(
("partition_key", "schedule", "should_raise"),
@@ -4383,6 +4415,92 @@ def test_disable_bundle_versioning(disable,
bundle_version, expected, dag_maker,
assert dr.bundle_version == expected
+def test_create_dagrun_uses_resolved_bundle_version_for_integrity(dag_maker,
session, clear_dags):
+ """
+ When no explicit bundle_version is passed, the live dag drives TI creation
and
+ created_dag_version points to the latest serialized version.
DagRun.bundle_version
+ still records the DagModel.bundle_version for auditing purposes.
+ """
+ with dag_maker(
+ dag_id="test_dag_bundle_version_integrity",
+ session=session,
+ serialized=True,
+ bundle_version="v1",
+ ) as _dag_v1:
+ EmptyOperator(task_id="t1")
+
+ with dag_maker(
+ dag_id="test_dag_bundle_version_integrity",
+ session=session,
+ serialized=True,
+ bundle_version="v2",
+ ) as dag_v2:
+ EmptyOperator(task_id="t1")
+ EmptyOperator(task_id="t2")
+
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id ==
dag_v2.dag_id))
+ dag_model.bundle_version = "v1"
+ session.commit()
+
+ dr = dag_v2.create_dagrun(
+ run_id="bundle_version_integrity",
+ run_after=pendulum.now(),
+ run_type="manual",
+ triggered_by=DagRunTriggeredByType.TEST,
+ state=None,
+ )
+
+ # DagRun.bundle_version records the DagModel value at trigger time (audit
field).
+ assert dr.bundle_version == "v1"
+ # created_dag_version reflects the latest serialized version (v2), not the
DagModel audit value.
+ assert dr.created_dag_version.bundle_version == "v2"
+ # TIs come from the live dag (dag_v2 with t1+t2), not from the old
serialized version.
+ assert {ti.task_id for ti in dr.get_task_instances(session=session)} ==
{"t1", "t2"}
+
+
+def test_create_dagrun_without_bundle_version_uses_live_dag(dag_maker,
session, clear_dags):
+ """
+ When no explicit bundle_version is passed, TIs are created from the live
dag even if
+ DagModel.bundle_version points to an older version. This confirms
backfills and other
+ callers that don't pass bundle_version are unaffected by the
bundle_version feature.
+ """
+ with dag_maker(
+ dag_id="test_dag_backfill_bundle_version",
+ session=session,
+ serialized=True,
+ bundle_version="v1",
+ ) as _dag_v1:
+ EmptyOperator(task_id="t1")
+
+ with dag_maker(
+ dag_id="test_dag_backfill_bundle_version",
+ session=session,
+ serialized=True,
+ bundle_version="v2",
+ ) as dag_v2:
+ EmptyOperator(task_id="t1")
+ EmptyOperator(task_id="t2")
+
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id ==
dag_v2.dag_id))
+ dag_model.bundle_version = "v1"
+ session.commit()
+
+ dr = dag_v2.create_dagrun(
+ run_id="no_bundle_version_uses_live_dag",
+ run_after=pendulum.now(),
+ run_type="manual",
+ triggered_by=DagRunTriggeredByType.TEST,
+ state=None,
+ )
+
+ # TIs come from the live dag (dag_v2), not from the v1 serialized version.
+ assert {ti.task_id for ti in dr.get_task_instances(session=session)} ==
{"t1", "t2"}
+ # created_dag_version reflects the latest serialization (v2).
+ assert dr.created_dag_version.bundle_version == "v2"
+ # DagRun.bundle_version still records the DagModel value at trigger time.
+ assert dr.bundle_version == "v1"
+
+
def test_get_run_data_interval():
with DAG("dag", schedule=None, start_date=DEFAULT_DATE) as dag:
EmptyOperator(task_id="empty_task")
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index d4e39a8ad19..9bb75d8dd87 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -830,6 +830,7 @@ class MaterializeAssetBody(BaseModel):
conf: Annotated[dict[str, Any] | None, Field(title="Conf")] = None
note: Annotated[str | None, Field(title="Note")] = None
partition_key: Annotated[str | None, Field(title="Partition Key")] = None
+ bundle_version: Annotated[str | None, Field(title="Bundle Version")] = None
class NewTaskResponse(BaseModel):
@@ -1160,6 +1161,7 @@ class TriggerDAGRunPostBody(BaseModel):
conf: Annotated[dict[str, Any] | None, Field(title="Conf")] = None
note: Annotated[str | None, Field(title="Note")] = None
partition_key: Annotated[str | None, Field(title="Partition Key")] = None
+ bundle_version: Annotated[str | None, Field(title="Bundle Version")] = None
class TriggerResponse(BaseModel):
diff --git a/devel-common/src/tests_common/test_utils/dag.py
b/devel-common/src/tests_common/test_utils/dag.py
index 6e02ddf61ab..891176eb499 100644
--- a/devel-common/src/tests_common/test_utils/dag.py
+++ b/devel-common/src/tests_common/test_utils/dag.py
@@ -41,15 +41,17 @@ def create_scheduler_dag(dag: DAG | SerializedDAG) ->
SerializedDAG:
def sync_dag_to_db(
dag: DAG,
bundle_name: str = "testing",
+ bundle_version: str | None = None,
session: Session = NEW_SESSION,
) -> SerializedDAG:
- return sync_dags_to_db([dag], bundle_name=bundle_name, session=session)[0]
+ return sync_dags_to_db([dag], bundle_name=bundle_name,
bundle_version=bundle_version, session=session)[0]
@provide_session
def sync_dags_to_db(
dags: Collection[DAG],
bundle_name: str = "testing",
+ bundle_version: str | None = None,
session: Session = NEW_SESSION,
) -> Sequence[SerializedDAG]:
"""
@@ -68,10 +70,12 @@ def sync_dags_to_db(
def _write_dag(dag: DAG) -> SerializedDAG:
data = DagSerialization.to_dict(dag)
- SerializedDagModel.write_dag(LazyDeserializedDAG(data=data),
bundle_name, session=session)
+ SerializedDagModel.write_dag(
+ LazyDeserializedDAG(data=data), bundle_name, bundle_version,
session=session
+ )
return DagSerialization.from_dict(data)
- SerializedDAG.bulk_write_to_db(bundle_name, None, dags, session=session)
+ SerializedDAG.bulk_write_to_db(bundle_name, bundle_version, dags,
session=session)
scheduler_dags = [_write_dag(dag) for dag in dags]
session.flush()
return scheduler_dags
