This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new fce31dd2c95 [v3-3-test] Rename PartitionAtRuntime as
PartitionedAtRuntime for consistency (#68978) (#68987)
fce31dd2c95 is described below
commit fce31dd2c952964a2303d2554d93ec28b92be033
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jun 26 08:56:13 2026 +0800
[v3-3-test] Rename PartitionAtRuntime as PartitionedAtRuntime for
consistency (#68978) (#68987)
Co-authored-by: Wei Lee <[email protected]>
---
.../docs/authoring-and-scheduling/assets.rst | 12 +++++-----
.../example_dags/example_asset_partition.py | 12 +++++-----
airflow-core/src/airflow/serialization/encoders.py | 8 +++----
airflow-core/src/airflow/timetables/base.py | 2 +-
airflow-core/src/airflow/timetables/simple.py | 6 ++---
.../tests/unit/api/common/test_trigger_dag.py | 12 +++++-----
.../core_api/routes/public/test_assets.py | 6 ++---
.../core_api/routes/public/test_dag_run.py | 26 +++++++++++-----------
airflow-core/tests/unit/assets/test_manager.py | 4 ++--
.../tests/unit/dag_processing/test_collection.py | 4 ++--
airflow-core/tests/unit/jobs/test_scheduler_job.py | 6 ++---
airflow-core/tests/unit/models/test_dag.py | 6 ++---
.../tests/unit/models/test_taskinstance.py | 20 ++++++++---------
.../unit/serialization/test_dag_serialization.py | 10 ++++-----
task-sdk/docs/api.rst | 2 +-
task-sdk/src/airflow/sdk/__init__.py | 6 ++---
task-sdk/src/airflow/sdk/__init__.pyi | 4 ++--
task-sdk/src/airflow/sdk/bases/timetable.py | 2 +-
.../airflow/sdk/definitions/timetables/assets.py | 2 +-
task-sdk/tests/task_sdk/definitions/test_dag.py | 17 ++++++++++----
20 files changed, 88 insertions(+), 79 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/assets.rst
b/airflow-core/docs/authoring-and-scheduling/assets.rst
index d039fbfd1ff..a74664302da 100644
--- a/airflow-core/docs/authoring-and-scheduling/assets.rst
+++ b/airflow-core/docs/authoring-and-scheduling/assets.rst
@@ -781,7 +781,7 @@ semantics (the default).
DAG,
Asset,
FixedKeyMapper,
- PartitionAtRuntime,
+ PartitionedAtRuntime,
PartitionedAssetTimetable,
RollupMapper,
SegmentWindow,
@@ -792,7 +792,7 @@ semantics (the default).
@asset(
uri="file://incoming/player-stats/multi-region.csv",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
)
def multi_region_player_stats(self, outlet_events):
# Emit one event per region in a single run.
@@ -836,17 +836,17 @@ Setting partition keys at runtime
When the partition key is not known ahead of time (for example, a watermark
discovered from the source data, a late-arriving file, or a backfill request),
let the producing task decide it while it runs. Schedule the producer with
-``PartitionAtRuntime()`` and record the key(s) on the emitted event with
+``PartitionedAtRuntime()`` and record the key(s) on the emitted event with
``outlet_events[self].add_partitions(...)``:
.. code-block:: python
- from airflow.sdk import PartitionAtRuntime, asset
+ from airflow.sdk import PartitionedAtRuntime, asset
@asset(
uri="file://incoming/player-stats/live-region.csv",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
)
def live_region_player_stats(self, outlet_events):
# The key is only known once the task runs.
@@ -862,7 +862,7 @@ keys collapse to a single event:
@asset(
uri="file://incoming/player-stats/multi-region.csv",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
)
def multi_region_player_stats(self, outlet_events):
outlet_events[self].add_partitions(["us", "eu", "apac"])
diff --git a/airflow-core/src/airflow/example_dags/example_asset_partition.py
b/airflow-core/src/airflow/example_dags/example_asset_partition.py
index a7929ed9258..f4171bb46f1 100644
--- a/airflow-core/src/airflow/example_dags/example_asset_partition.py
+++ b/airflow-core/src/airflow/example_dags/example_asset_partition.py
@@ -30,8 +30,8 @@ from airflow.sdk import (
IdentityMapper,
MinimumCount,
MonthWindow,
- PartitionAtRuntime,
PartitionedAssetTimetable,
+ PartitionedAtRuntime,
ProductMapper,
RollupMapper,
SegmentWindow,
@@ -160,7 +160,7 @@ regional_sales =
Asset(uri="file://incoming/sales/regional.csv", name="regional_
with DAG(
dag_id="ingest_regional_sales",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
tags=["example", "sales", "ingestion"],
):
"""Produce regional sales data with composite ``region|timestamp``
partition keys at runtime."""
@@ -207,7 +207,7 @@ region_raw_stats =
Asset(uri="file://incoming/player-stats/by-region.csv", name=
with DAG(
dag_id="ingest_region_stats",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
tags=["example", "player-stats", "regional"],
):
"""
@@ -244,14 +244,14 @@ def regional_stats_breakdown():
@asset(
uri="file://incoming/player-stats/live-region.csv",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
tags=["player-stats", "runtime"],
)
def live_region_player_stats(self, outlet_events):
"""
Produce a single region partition whose key is decided at runtime.
- This asset demonstrates PartitionAtRuntime, which records the partition
key on the
+ This asset demonstrates PartitionedAtRuntime, which records the partition
key on the
emitted event with ``add_partitions`` while the task runs rather than from
a timetable.
"""
outlet_events[self].add_partitions("us")
@@ -281,7 +281,7 @@ with DAG(
@asset(
uri="file://incoming/player-stats/multi-region.csv",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
tags=["player-stats", "runtime"],
)
def multi_region_player_stats(self, outlet_events):
diff --git a/airflow-core/src/airflow/serialization/encoders.py
b/airflow-core/src/airflow/serialization/encoders.py
index b31b7c828c6..8ab533b888a 100644
--- a/airflow-core/src/airflow/serialization/encoders.py
+++ b/airflow-core/src/airflow/serialization/encoders.py
@@ -70,8 +70,8 @@ from airflow.sdk.bases.timetable import BaseTimetable
from airflow.sdk.definitions.asset import AssetRef
from airflow.sdk.definitions.timetables.assets import (
AssetTriggeredTimetable,
- PartitionAtRuntime,
PartitionedAssetTimetable,
+ PartitionedAtRuntime,
)
from airflow.sdk.definitions.timetables.simple import ContinuousTimetable,
NullTimetable, OnceTimetable
from airflow.sdk.definitions.timetables.trigger import CronPartitionTimetable
@@ -329,7 +329,7 @@ class _Serializer:
MultipleCronTriggerTimetable:
"airflow.timetables.trigger.MultipleCronTriggerTimetable",
NullTimetable: "airflow.timetables.simple.NullTimetable",
OnceTimetable: "airflow.timetables.simple.OnceTimetable",
- PartitionAtRuntime: "airflow.timetables.simple.PartitionAtRuntime",
+ PartitionedAtRuntime: "airflow.timetables.simple.PartitionedAtRuntime",
PartitionedAssetTimetable:
"airflow.timetables.simple.PartitionedAssetTimetable",
}
@@ -355,9 +355,9 @@ class _Serializer:
@serialize_timetable.register(ContinuousTimetable)
@serialize_timetable.register(NullTimetable)
@serialize_timetable.register(OnceTimetable)
- @serialize_timetable.register(PartitionAtRuntime)
+ @serialize_timetable.register(PartitionedAtRuntime)
def _(
- self, timetable: ContinuousTimetable | NullTimetable | OnceTimetable |
PartitionAtRuntime
+ self, timetable: ContinuousTimetable | NullTimetable | OnceTimetable |
PartitionedAtRuntime
) -> dict[str, Any]:
return {}
diff --git a/airflow-core/src/airflow/timetables/base.py
b/airflow-core/src/airflow/timetables/base.py
index 4bee4ab38a3..84d8aa2e46f 100644
--- a/airflow-core/src/airflow/timetables/base.py
+++ b/airflow-core/src/airflow/timetables/base.py
@@ -240,7 +240,7 @@ class Timetable(Protocol):
partitioned_at_runtime: bool = False
"""Whether this timetable defers partition selection to task runtime.
- *True* for :class:`~airflow.timetables.simple.PartitionAtRuntime`;
+ *True* for :class:`~airflow.timetables.simple.PartitionedAtRuntime`;
downstream code can branch on this flag instead of using ``isinstance``.
"""
diff --git a/airflow-core/src/airflow/timetables/simple.py
b/airflow-core/src/airflow/timetables/simple.py
index a8b499dbe50..87741a8eb0d 100644
--- a/airflow-core/src/airflow/timetables/simple.py
+++ b/airflow-core/src/airflow/timetables/simple.py
@@ -186,11 +186,11 @@ class ContinuousTimetable(_TrivialTimetable):
return DagRunInfo.interval(start, end)
-class PartitionAtRuntime(NullTimetable):
+class PartitionedAtRuntime(NullTimetable):
"""
Timetable that never schedules anything; partition keys are set at runtime.
- This corresponds to ``schedule=PartitionAtRuntime()``.
+ This corresponds to ``schedule=PartitionedAtRuntime()``.
A run's ``partition_key`` (run-level provenance) must be supplied at
trigger
time — for example via the REST API's ``partition_key`` field. Partition
keys
@@ -204,7 +204,7 @@ class PartitionAtRuntime(NullTimetable):
@property
def summary(self) -> str:
- return "PartitionAtRuntime"
+ return "PartitionedAtRuntime"
class AssetTriggeredTimetable(_TrivialTimetable):
diff --git a/airflow-core/tests/unit/api/common/test_trigger_dag.py
b/airflow-core/tests/unit/api/common/test_trigger_dag.py
index 6fc48bd0814..e1837426bc1 100644
--- a/airflow-core/tests/unit/api/common/test_trigger_dag.py
+++ b/airflow-core/tests/unit/api/common/test_trigger_dag.py
@@ -25,7 +25,7 @@ from airflow.api.common.trigger_dag import trigger_dag
from airflow.exceptions import InvalidPartitionKeyError
from airflow.models import DagModel
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.timetables.simple import PartitionAtRuntime
+from airflow.timetables.simple import PartitionedAtRuntime
from airflow.timetables.trigger import CronPartitionTimetable
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -129,18 +129,18 @@ def
test_trigger_dag_raises_invalid_partition_key_for_cron_partition_timetable(d
)
-def
test_trigger_dag_partition_at_runtime_leaves_partition_date_none(dag_maker,
session):
- """PartitionAtRuntime Dags accept arbitrary keys; partition_date stays
None."""
+def
test_trigger_dag_partitioned_at_runtime_leaves_partition_date_none(dag_maker,
session):
+ """PartitionedAtRuntime Dags accept arbitrary keys; partition_date stays
None."""
with dag_maker(
session=session,
- dag_id="TEST_PARTITION_AT_RUNTIME",
- schedule=PartitionAtRuntime(),
+ dag_id="TEST_PARTITIONED_AT_RUNTIME",
+ schedule=PartitionedAtRuntime(),
):
EmptyOperator(task_id="mytask")
session.commit()
dag_run = trigger_dag(
- dag_id="TEST_PARTITION_AT_RUNTIME",
+ dag_id="TEST_PARTITIONED_AT_RUNTIME",
triggered_by=DagRunTriggeredByType.REST_API,
partition_key="arbitrary-runtime-key",
session=session,
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 b65036b72d9..33c43785a55 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
@@ -41,7 +41,7 @@ from airflow.models.dagrun import DagRun
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.timetables.simple import PartitionAtRuntime
+from airflow.timetables.simple import PartitionedAtRuntime
from airflow.utils.session import provide_session
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunType
@@ -1519,9 +1519,9 @@ class TestPostAssetMaterialize(TestAssets):
i: am.to_serialized() for i, am in
enumerate(self.create_assets(session=session, num=3), start=1)
}
# DAG_ASSET1_ID is materialized with a partition_key in several tests
below, so it must be a
- # partitioned Dag. PartitionAtRuntime accepts runtime-discovered
partition keys without
+ # partitioned Dag. PartitionedAtRuntime accepts runtime-discovered
partition keys without
# requiring a partitioned timetable.
- with dag_maker(self.DAG_ASSET1_ID, schedule=PartitionAtRuntime(),
session=session):
+ with dag_maker(self.DAG_ASSET1_ID, schedule=PartitionedAtRuntime(),
session=session):
EmptyOperator(task_id="task", outlets=assets[1])
with dag_maker(self.DAG_ASSET2_ID_A, schedule=None, session=session):
EmptyOperator(task_id="task", outlets=assets[2])
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 5b5a0293ffc..39a391f80f1 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
@@ -45,7 +45,7 @@ from airflow.providers.standard.operators.empty import
EmptyOperator
from airflow.sdk import Asset, Param, result, task
from airflow.settings import _configure_async_session
from airflow.timetables.interval import CronDataIntervalTimetable
-from airflow.timetables.simple import PartitionAtRuntime,
PartitionedAssetTimetable
+from airflow.timetables.simple import PartitionedAssetTimetable,
PartitionedAtRuntime
from airflow.timetables.trigger import CronPartitionTimetable
from airflow.utils.session import provide_session
from airflow.utils.state import DagRunState, State
@@ -1619,8 +1619,8 @@ class TestGetDagRunAssetTriggerEvents:
def test_should_respond_200(self, partition_key, test_client, dag_maker,
session):
asset1 = Asset(name="ds1", uri="file:///da1")
- # Use PartitionAtRuntime for partitioned cases so the partition_key
gate does not reject the key.
- source_schedule = PartitionAtRuntime() if partition_key is not None
else timedelta(days=1)
+ # Use PartitionedAtRuntime for partitioned cases so the partition_key
gate does not reject the key.
+ source_schedule = PartitionedAtRuntime() if partition_key is not None
else timedelta(days=1)
with dag_maker(
dag_id="source_dag", start_date=START_DATE1,
schedule=source_schedule, session=session
):
@@ -1639,7 +1639,7 @@ class TestGetDagRunAssetTriggerEvents:
)
session.add(event)
- trigger_schedule = PartitionAtRuntime() if partition_key is not None
else timedelta(days=1)
+ trigger_schedule = PartitionedAtRuntime() if partition_key is not None
else timedelta(days=1)
with dag_maker(
dag_id="TEST_DAG_ID", start_date=START_DATE1,
schedule=trigger_schedule, session=session
):
@@ -1650,7 +1650,7 @@ class TestGetDagRunAssetTriggerEvents:
"partition_key": partition_key,
}
if partition_key is not None:
- # PartitionAtRuntime is a null-timetable with no scheduled runs;
supply logical_date=None
+ # PartitionedAtRuntime is a null-timetable with no scheduled runs;
supply logical_date=None
# 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)
@@ -3503,18 +3503,18 @@ class TestTriggerDagRun:
assert response.status_code == 200
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
- def
test_should_respond_200_when_partition_key_given_for_partition_at_runtime_dag(
+ def
test_should_respond_200_when_partition_key_given_for_partitioned_at_runtime_dag(
self, dag_maker, test_client, session
):
- """partition_key on a PartitionAtRuntime Dag must also be accepted
(deferred validation).
+ """partition_key on a PartitionedAtRuntime Dag must also be accepted
(deferred validation).
partitioned_at_runtime=True means the Dag accepts runtime-discovered
partition keys, so
the REST layer must not reject it even though timetable.partitioned is
False.
"""
- runtime_dag_id = "test_partition_at_runtime_dag_trigger"
+ runtime_dag_id = "test_partitioned_at_runtime_dag_trigger"
with dag_maker(
dag_id=runtime_dag_id,
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
start_date=START_DATE1,
session=session,
serialized=True,
@@ -3638,14 +3638,14 @@ class TestTriggerDagRun:
assert response.status_code == 400
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
- def test_trigger_partition_at_runtime_dag_leaves_partition_date_none(
+ def test_trigger_partitioned_at_runtime_dag_leaves_partition_date_none(
self, dag_maker, test_client, session
):
- """PartitionAtRuntime Dag with an arbitrary key must produce
partition_date=None."""
- runtime_dag_id = "test_trigger_partition_at_runtime_none_date"
+ """PartitionedAtRuntime Dag with an arbitrary key must produce
partition_date=None."""
+ runtime_dag_id = "test_trigger_partitioned_at_runtime_none_date"
with dag_maker(
dag_id=runtime_dag_id,
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
start_date=START_DATE1,
session=session,
serialized=True,
diff --git a/airflow-core/tests/unit/assets/test_manager.py
b/airflow-core/tests/unit/assets/test_manager.py
index 54ddd66a2d4..b788b9ab286 100644
--- a/airflow-core/tests/unit/assets/test_manager.py
+++ b/airflow-core/tests/unit/assets/test_manager.py
@@ -494,10 +494,10 @@ class TestAssetManager:
# partition-at-runtime Dag so its run can carry a ``partition_key``
that
# the emitted ``AssetEvent`` inherits.
from airflow.models.taskinstance import TaskInstance
- from airflow.sdk import PartitionAtRuntime
+ from airflow.sdk import PartitionedAtRuntime
with dag_maker(
- dag_id="stamp-producer", schedule=PartitionAtRuntime(),
session=session
+ dag_id="stamp-producer", schedule=PartitionedAtRuntime(),
session=session
) as producer_dag:
from airflow.providers.standard.operators.empty import
EmptyOperator
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index fe15045598f..ee8d7fa0fcd 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -64,7 +64,7 @@ from airflow.sdk.definitions.timetables.assets import
PartitionedAssetTimetable
from airflow.serialization.definitions.assets import SerializedAsset
from airflow.serialization.encoders import encode_trigger,
ensure_serialized_asset
from airflow.serialization.serialized_objects import LazyDeserializedDAG
-from airflow.timetables.simple import PartitionAtRuntime
+from airflow.timetables.simple import PartitionedAtRuntime
from airflow.triggers.base import BaseEventTrigger
from airflow.utils.types import DagRunType
@@ -136,7 +136,7 @@ def
test_statement_latest_runs_loads_timetable_fields(dag_maker, session):
@pytest.mark.db_test
def test_statement_latest_runs_partitioned_sorted_by_partition_date(dag_maker,
session):
- with dag_maker("fake-dag", schedule=PartitionAtRuntime()):
+ with dag_maker("fake-dag", schedule=PartitionedAtRuntime()):
pass
dag_maker.sync_dagbag_to_db()
for i, (run_id, partition_key, partition_date) in enumerate(
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 3b950f71d5b..e2d5977eb86 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -132,8 +132,8 @@ from airflow.serialization.encoders import
ensure_serialized_asset
from airflow.serialization.serialized_objects import LazyDeserializedDAG
from airflow.timetables.base import DagRunInfo, DataInterval,
compute_rollup_fingerprint
from airflow.timetables.simple import (
- PartitionAtRuntime,
PartitionedAssetTimetable as CorePartitionedAssetTimetable,
+ PartitionedAtRuntime,
)
from airflow.utils.session import NEW_SESSION, create_session, provide_session
from airflow.utils.sqlalchemy import with_row_locks
@@ -10255,7 +10255,7 @@ def _produce_and_register_asset_event(
if expected_partition_key is None:
expected_partition_key = partition_key
- with dag_maker(dag_id=dag_id, schedule=PartitionAtRuntime(),
session=session) as dag:
+ with dag_maker(dag_id=dag_id, schedule=PartitionedAtRuntime(),
session=session) as dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(
@@ -10707,7 +10707,7 @@ def
test_consumer_dag_run_partition_date_is_none_when_task_key_diverges(
with dag_maker(
dag_id="asset-event-producer",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
session=session,
) as dag:
EmptyOperator(task_id="hi", outlets=[asset_1])
diff --git a/airflow-core/tests/unit/models/test_dag.py
b/airflow-core/tests/unit/models/test_dag.py
index 090cd2d4756..ad4ed2c134d 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -71,7 +71,7 @@ from airflow.sdk import (
DAG,
BaseOperator,
CronPartitionTimetable,
- PartitionAtRuntime,
+ PartitionedAtRuntime,
TaskGroup,
setup,
task as task_decorator,
@@ -1559,7 +1559,7 @@ class TestDag:
"""DagRun-level type check rejects int partition_key even for
partitioned Dags."""
with dag_maker(
"test_create_dagrun_partitioned_int_key",
- schedule=PartitionAtRuntime(),
+ schedule=PartitionedAtRuntime(),
):
...
with pytest.raises(
@@ -1582,7 +1582,7 @@ class TestDag:
("my-key", None, True),
(None, None, False),
("my-key", CronPartitionTimetable("@daily", timezone="UTC"),
False),
- ("my-key", PartitionAtRuntime(), False),
+ ("my-key", PartitionedAtRuntime(), False),
],
)
def test_serialized_dag_validate_partition_key(
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 97ec0242235..89ac6b12048 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -106,7 +106,7 @@ from airflow.ti_deps.dependencies_states import
RUNNABLE_STATES
from airflow.ti_deps.deps.base_ti_dep import TIDepStatus
from airflow.ti_deps.deps.ready_to_reschedule import ReadyToRescheduleDep
from airflow.ti_deps.deps.trigger_rule_dep import TriggerRuleDep,
_UpstreamTIStates
-from airflow.timetables.simple import PartitionAtRuntime
+from airflow.timetables.simple import PartitionedAtRuntime
from airflow.utils.session import create_session, provide_session
from airflow.utils.span_status import SpanStatus
from airflow.utils.state import DagRunState, State, TaskInstanceState
@@ -3568,7 +3568,7 @@ def
test_register_asset_changes_in_db_no_outlets_is_a_noop(dag_maker, session):
def test_when_dag_run_has_partition_then_asset_does(dag_maker, session):
asset = Asset(name="hello")
- with dag_maker(dag_id="asset_event_tester", schedule=PartitionAtRuntime())
as dag:
+ with dag_maker(dag_id="asset_event_tester",
schedule=PartitionedAtRuntime()) as dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(partition_key="abc123", session=session)
assert dr.partition_key == "abc123"
@@ -3591,7 +3591,7 @@ def
test_when_dag_run_has_partition_and_downstreams_listening_then_tables_popula
session,
):
asset = Asset(name="hello")
- with dag_maker(dag_id="asset_event_tester", schedule=PartitionAtRuntime(),
session=session) as dag:
+ with dag_maker(dag_id="asset_event_tester",
schedule=PartitionedAtRuntime(), session=session) as dag:
EmptyOperator(task_id="hi", outlets=[asset])
dag1_id = dag.dag_id
dr = dag_maker.create_dagrun(partition_key="abc123", session=session)
@@ -3657,7 +3657,7 @@ def
test_runtime_partition_key_does_not_backfill_dag_run_when_none(dag_maker, se
def
test_runtime_partition_key_does_not_overwrite_scheduler_partition(dag_maker,
session):
"""Task-emitted key lands on the AssetEvent but does NOT overwrite a
scheduler-set DagRun.partition_key."""
asset = Asset(name="hello")
- with dag_maker(dag_id="rt_pk_no_overwrite", schedule=PartitionAtRuntime())
as dag:
+ with dag_maker(dag_id="rt_pk_no_overwrite",
schedule=PartitionedAtRuntime()) as dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(partition_key="scheduler-key",
session=session)
[ti] = dr.get_task_instances(session=session)
@@ -3708,7 +3708,7 @@ def
test_runtime_partition_key_inherits_dag_run_key_when_event_has_no_key(dag_ma
can still be routed correctly.
"""
asset = Asset(name="hello")
- with dag_maker(dag_id="rt_pk_none", schedule=PartitionAtRuntime()) as dag:
+ with dag_maker(dag_id="rt_pk_none", schedule=PartitionedAtRuntime()) as
dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(partition_key="from-run", session=session)
[ti] = dr.get_task_instances(session=session)
@@ -3728,7 +3728,7 @@ def
test_runtime_partition_key_inherits_dag_run_key_when_event_has_no_key(dag_ma
def test_runtime_partition_key_mixed_events_for_same_asset(dag_maker, session):
"""One event with an explicit key + one without produce two AssetEvents
(explicit key + inherited run key)."""
asset = Asset(name="hello")
- with dag_maker(dag_id="rt_pk_mixed", schedule=PartitionAtRuntime()) as dag:
+ with dag_maker(dag_id="rt_pk_mixed", schedule=PartitionedAtRuntime()) as
dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(partition_key="from-run", session=session)
[ti] = dr.get_task_instances(session=session)
@@ -3773,17 +3773,17 @@ def
test_runtime_partition_key_event_stays_none_when_no_key_and_no_run_key(dag_m
assert event.partition_key is None
-def
test_runtime_partition_key_does_not_backfill_partition_at_runtime_run(dag_maker,
session):
- """Task-emitted key lands on AssetEvent but does NOT back-fill
DagRun.partition_key on a PartitionAtRuntime run.
+def
test_runtime_partition_key_does_not_backfill_partitioned_at_runtime_run(dag_maker,
session):
+ """Task-emitted key lands on AssetEvent but does NOT back-fill
DagRun.partition_key on a PartitionedAtRuntime run.
Provenance contract: DagRun.partition_key is set only at
run-creation/trigger time.
- A PartitionAtRuntime Dag triggered via REST without a partition_key starts
with
+ A PartitionedAtRuntime Dag triggered via REST without a partition_key
starts with
partition_key=None. Even when a task discovers a partition at runtime and
emits an
outlet event carrying an explicit key, DagRun.partition_key must remain
None — the
key belongs to the AssetEvent, not to the run's provenance.
"""
asset = Asset(name="hello")
- with dag_maker(dag_id="rt_pk_par_backfill", schedule=PartitionAtRuntime())
as dag:
+ with dag_maker(dag_id="rt_pk_par_backfill",
schedule=PartitionedAtRuntime()) as dag:
EmptyOperator(task_id="hi", outlets=[asset])
dr = dag_maker.create_dagrun(session=session)
assert dr.partition_key is None
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py
b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 61a66c60fb6..07459a20117 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -93,7 +93,7 @@ from airflow.task.priority_strategy import (
validate_and_load_priority_weight_strategy,
)
from airflow.ti_deps.deps.ready_to_reschedule import ReadyToRescheduleDep
-from airflow.timetables.simple import NullTimetable, OnceTimetable,
PartitionAtRuntime
+from airflow.timetables.simple import NullTimetable, OnceTimetable,
PartitionedAtRuntime
from airflow.timetables.trigger import CronPartitionTimetable
from airflow.triggers.base import StartTriggerArgs
from airflow.utils.types import DagRunType
@@ -2778,11 +2778,11 @@ class TestStringifiedDAGs:
assert dr.partition_key == "2025-01-01T00:00:00"
@pytest.mark.db_test
- def
test_create_dagrun_accepts_partition_key_for_partition_at_runtime_dag(self,
dag_maker):
- """create_dagrun does not raise when partition_key is passed to a
PartitionAtRuntime Dag."""
+ def
test_create_dagrun_accepts_partition_key_for_partitioned_at_runtime_dag(self,
dag_maker):
+ """create_dagrun does not raise when partition_key is passed to a
PartitionedAtRuntime Dag."""
with dag_maker(
- dag_id="test_partition_at_runtime",
- schedule=PartitionAtRuntime(),
+ dag_id="test_partitioned_at_runtime",
+ schedule=PartitionedAtRuntime(),
serialized=True,
):
pass
diff --git a/task-sdk/docs/api.rst b/task-sdk/docs/api.rst
index d31695f486e..984b0cd7261 100644
--- a/task-sdk/docs/api.rst
+++ b/task-sdk/docs/api.rst
@@ -214,7 +214,7 @@ Timetables
.. autoapiclass:: airflow.sdk.MultipleCronTriggerTimetable
-.. autoapiclass:: airflow.sdk.PartitionAtRuntime
+.. autoapiclass:: airflow.sdk.PartitionedAtRuntime
.. autoapiclass:: airflow.sdk.PartitionedAssetTimetable
diff --git a/task-sdk/src/airflow/sdk/__init__.py
b/task-sdk/src/airflow/sdk/__init__.py
index 8e8953a8aca..f3452389d17 100644
--- a/task-sdk/src/airflow/sdk/__init__.py
+++ b/task-sdk/src/airflow/sdk/__init__.py
@@ -67,7 +67,7 @@ __all__ = [
"ObjectStoragePath",
"Param",
"ParamsDict",
- "PartitionAtRuntime",
+ "PartitionedAtRuntime",
"PartitionedAssetTimetable",
"PartitionMapper",
"PokeReturnValue",
@@ -199,8 +199,8 @@ if TYPE_CHECKING:
from airflow.sdk.definitions.template import literal
from airflow.sdk.definitions.timetables.assets import (
AssetOrTimeSchedule,
- PartitionAtRuntime,
PartitionedAssetTimetable,
+ PartitionedAtRuntime,
)
from airflow.sdk.definitions.timetables.events import EventsTimetable
from airflow.sdk.definitions.timetables.interval import (
@@ -269,7 +269,7 @@ __lazy_imports: dict[str, str] = {
"ObjectStoragePath": ".io.path",
"Param": ".definitions.param",
"ParamsDict": ".definitions.param",
- "PartitionAtRuntime": ".definitions.timetables.assets",
+ "PartitionedAtRuntime": ".definitions.timetables.assets",
"PartitionedAssetTimetable": ".definitions.timetables.assets",
"PartitionMapper": ".definitions.partition_mappers.base",
"PokeReturnValue": ".bases.sensor",
diff --git a/task-sdk/src/airflow/sdk/__init__.pyi
b/task-sdk/src/airflow/sdk/__init__.pyi
index 273b0579f3e..d48bf614407 100644
--- a/task-sdk/src/airflow/sdk/__init__.pyi
+++ b/task-sdk/src/airflow/sdk/__init__.pyi
@@ -110,8 +110,8 @@ from airflow.sdk.definitions.taskgroup import TaskGroup as
TaskGroup
from airflow.sdk.definitions.template import literal as literal
from airflow.sdk.definitions.timetables.assets import (
AssetOrTimeSchedule,
- PartitionAtRuntime,
PartitionedAssetTimetable,
+ PartitionedAtRuntime,
)
from airflow.sdk.definitions.timetables.events import EventsTimetable
from airflow.sdk.definitions.timetables.interval import (
@@ -177,7 +177,7 @@ __all__ = [
"ObjectStoragePath",
"Param",
"PokeReturnValue",
- "PartitionAtRuntime",
+ "PartitionedAtRuntime",
"PartitionedAssetTimetable",
"PartitionMapper",
"ProductMapper",
diff --git a/task-sdk/src/airflow/sdk/bases/timetable.py
b/task-sdk/src/airflow/sdk/bases/timetable.py
index bd37a6a0a79..27769d6b31c 100644
--- a/task-sdk/src/airflow/sdk/bases/timetable.py
+++ b/task-sdk/src/airflow/sdk/bases/timetable.py
@@ -51,7 +51,7 @@ class BaseTimetable:
"""
Whether this timetable defers partition selection to task runtime.
- *True* for :class:`~airflow.sdk.PartitionAtRuntime`; downstream code can
+ *True* for :class:`~airflow.sdk.PartitionedAtRuntime`; downstream code can
branch on this flag instead of using ``isinstance``.
"""
diff --git a/task-sdk/src/airflow/sdk/definitions/timetables/assets.py
b/task-sdk/src/airflow/sdk/definitions/timetables/assets.py
index a0c64936925..22208693588 100644
--- a/task-sdk/src/airflow/sdk/definitions/timetables/assets.py
+++ b/task-sdk/src/airflow/sdk/definitions/timetables/assets.py
@@ -54,7 +54,7 @@ class PartitionedAssetTimetable(AssetTriggeredTimetable):
default_partition_mapper: PartitionMapper = IdentityMapper()
-class PartitionAtRuntime(BaseTimetable):
+class PartitionedAtRuntime(BaseTimetable):
"""Marker timetable indicating that partition key(s) are determined at
runtime."""
can_be_scheduled = False
diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py
b/task-sdk/tests/task_sdk/definitions/test_dag.py
index 0e9f6c78b38..b8f9d01b5b0 100644
--- a/task-sdk/tests/task_sdk/definitions/test_dag.py
+++ b/task-sdk/tests/task_sdk/definitions/test_dag.py
@@ -25,7 +25,16 @@ from unittest import mock
import pytest
-from airflow.sdk import DAG, Context, Label, Param, PartitionAtRuntime,
TaskGroup, dag as dag_decorator, task
+from airflow.sdk import (
+ DAG,
+ Context,
+ Label,
+ Param,
+ PartitionedAtRuntime,
+ TaskGroup,
+ dag as dag_decorator,
+ task,
+)
from airflow.sdk.bases.operator import BaseOperator
from airflow.sdk.bases.timetable import BaseTimetable
from airflow.sdk.definitions.param import DagParam, ParamsDict
@@ -438,8 +447,8 @@ class TestDag:
with pytest.raises(ValueError, match="ContinuousTimetable requires
max_active_runs <= 1"):
dag = DAG("continuous", start_date=DEFAULT_DATE,
schedule="@continuous", max_active_runs=25)
- def test_only_partition_at_runtime_has_partitioned_at_runtime_flag(self):
- """Regression guard: across every BaseTimetable subclass, only
PartitionAtRuntime sets partitioned_at_runtime=True."""
+ def test_only_partitioned_at_runtime_has_partitioned_at_runtime_flag(self):
+ """Regression guard: across every BaseTimetable subclass, only
PartitionedAtRuntime sets partitioned_at_runtime=True."""
def all_subclasses(cls):
for sub in cls.__subclasses__():
@@ -447,7 +456,7 @@ class TestDag:
yield from all_subclasses(sub)
flagged = {c for c in all_subclasses(BaseTimetable) if
c.partitioned_at_runtime}
- assert flagged == {PartitionAtRuntime}
+ assert flagged == {PartitionedAtRuntime}
def test_dag_add_task_checks_trigger_rule(self):
# A non fail stop dag should allow any trigger rule
