Skip to content

Commit 3bb2dd5

Browse files
Lee-WLee-W
authored
refactor(partition-mapper): rename ToXXXMapper to StartOfXXXMapper (#64160)
1 parent 1680d60 commit 3bb2dd5

13 files changed

Lines changed: 188 additions & 181 deletions

File tree

‎airflow-core/docs/authoring-and-scheduling/assets.rst‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -434,13 +434,13 @@ For downstream partition-aware scheduling, use ``PartitionedAssetTimetable``:
434434

435435
.. code-block:: python
436436
437-
from airflow.sdk import DAG, HourlyMapper, PartitionedAssetTimetable
437+
from airflow.sdk import DAG, StartOfHourMapper, PartitionedAssetTimetable
438438
439439
with DAG(
440440
dag_id="clean_and_combine_player_stats",
441441
schedule=PartitionedAssetTimetable(
442442
assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
443-
default_partition_mapper=HourlyMapper(),
443+
default_partition_mapper=StartOfHourMapper(),
444444
),
445445
catchup=False,
446446
):
@@ -458,17 +458,17 @@ Partition mappers define how upstream partition keys are transformed to the
458458
downstream Dag partition key:
459459

460460
* ``IdentityMapper`` keeps keys unchanged.
461-
* Temporal mappers such as ``HourlyMapper``, ``DailyMapper``, and
462-
``YearlyMapper`` normalize time keys to a chosen grain. For input key
461+
* Temporal mappers such as ``StartOfHourMapper``, ``StartOfDayMapper``, and
462+
``StartOfYearMapper`` normalize time keys to a chosen grain. For input key
463463
``2026-03-10T09:37:51``, the default outputs are:
464464

465-
* ``HourlyMapper`` -> ``2026-03-10T09``
466-
* ``DailyMapper`` -> ``2026-03-10``
467-
* ``YearlyMapper`` -> ``2026``
465+
* ``StartOfHourMapper`` -> ``2026-03-10T09``
466+
* ``StartOfDayMapper`` -> ``2026-03-10``
467+
* ``StartOfYearMapper`` -> ``2026``
468468
* ``ProductMapper`` maps composite keys segment-by-segment.
469469
It applies one mapper per segment and then rejoins the mapped segments.
470470
For example, with key ``us|2026-03-10T09:00:00``,
471-
``ProductMapper(IdentityMapper(), DailyMapper())`` produces
471+
``ProductMapper(IdentityMapper(), StartOfDayMapper())`` produces
472472
``us|2026-03-10``.
473473
* ``AllowedKeyMapper`` validates that keys are in a fixed allow-list and
474474
passes the key through unchanged if valid.
@@ -481,10 +481,10 @@ Example of per-asset mapper configuration and composite-key mapping:
481481
482482
from airflow.sdk import (
483483
Asset,
484-
DailyMapper,
485484
IdentityMapper,
486485
PartitionedAssetTimetable,
487486
ProductMapper,
487+
StartOfDayMapper,
488488
)
489489
490490
regional_sales = Asset(uri="file://incoming/sales/regional.csv", name="regional_sales")
@@ -493,7 +493,7 @@ Example of per-asset mapper configuration and composite-key mapping:
493493
dag_id="aggregate_regional_sales",
494494
schedule=PartitionedAssetTimetable(
495495
assets=regional_sales,
496-
default_partition_mapper=ProductMapper(IdentityMapper(), DailyMapper()),
496+
default_partition_mapper=ProductMapper(IdentityMapper(), StartOfDayMapper()),
497497
),
498498
):
499499
...
@@ -503,7 +503,7 @@ You can also override mappers for specific upstream assets with
503503

504504
.. code-block:: python
505505
506-
from airflow.sdk import Asset, DAG, DailyMapper, IdentityMapper, PartitionedAssetTimetable
506+
from airflow.sdk import Asset, DAG, StartOfDayMapper, IdentityMapper, PartitionedAssetTimetable
507507
508508
hourly_sales = Asset(uri="file://incoming/sales/hourly.csv", name="hourly_sales")
509509
daily_targets = Asset(uri="file://incoming/sales/targets.csv", name="daily_targets")
@@ -513,7 +513,7 @@ You can also override mappers for specific upstream assets with
513513
schedule=PartitionedAssetTimetable(
514514
assets=hourly_sales & daily_targets,
515515
# Default behavior: map timestamp-like keys to daily keys.
516-
default_partition_mapper=DailyMapper(),
516+
default_partition_mapper=StartOfDayMapper(),
517517
# Override for assets that already emit daily partition keys.
518518
partition_mapper_config={
519519
daily_targets: IdentityMapper(),

‎airflow-core/src/airflow/example_dags/example_asset_partition.py‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,9 @@
2727
IdentityMapper,
2828
PartitionedAssetTimetable,
2929
ProductMapper,
30-
ToDailyMapper,
31-
ToHourlyMapper,
32-
ToYearlyMapper,
30+
StartOfDayMapper,
31+
StartOfHourMapper,
32+
StartOfYearMapper,
3333
asset,
3434
task,
3535
)
@@ -77,15 +77,15 @@ def team_c_player_stats():
7777
dag_id="clean_and_combine_player_stats",
7878
schedule=PartitionedAssetTimetable(
7979
assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
80-
default_partition_mapper=ToHourlyMapper(),
80+
default_partition_mapper=StartOfHourMapper(),
8181
),
8282
catchup=False,
8383
tags=["player-stats", "cleanup"],
8484
):
8585
"""
8686
Combine hourly partitions from Team A, B and C into a single curated dataset.
8787
88-
This Dag demonstrates multi-asset partition alignment using ToHourlyMapper.
88+
This Dag demonstrates multi-asset partition alignment using StartOfHourMapper.
8989
"""
9090

9191
@task(outlets=[combined_player_stats])
@@ -101,7 +101,7 @@ def combine_player_stats(dag_run=None):
101101
@asset(
102102
uri="file://analytics/player-stats/computed-player-odds.csv",
103103
# Fallback to IdentityMapper if no partition_mapper is specified.
104-
# If we want to other temporal mapper (e.g., ToHourlyMapper) here,
104+
# If we want to other temporal mapper (e.g., StartOfHourMapper) here,
105105
# make sure the input_format is changed since the partition_key is now in "%Y-%m-%dT%H" format
106106
# instead of a valid timestamp
107107
schedule=PartitionedAssetTimetable(assets=combined_player_stats),
@@ -121,9 +121,9 @@ def compute_player_odds():
121121
schedule=PartitionedAssetTimetable(
122122
assets=(combined_player_stats & team_a_player_stats & Asset.ref(name="team_b_player_stats")),
123123
partition_mapper_config={
124-
combined_player_stats: ToYearlyMapper(), # incompatible on purpose
125-
team_a_player_stats: ToHourlyMapper(),
126-
Asset.ref(name="team_b_player_stats"): ToHourlyMapper(),
124+
combined_player_stats: StartOfYearMapper(), # incompatible on purpose
125+
team_a_player_stats: StartOfHourMapper(),
126+
Asset.ref(name="team_b_player_stats"): StartOfHourMapper(),
127127
},
128128
),
129129
catchup=False,
@@ -164,7 +164,7 @@ def ingest_sales():
164164
dag_id="aggregate_regional_sales",
165165
schedule=PartitionedAssetTimetable(
166166
assets=regional_sales,
167-
default_partition_mapper=ProductMapper(IdentityMapper(), ToDailyMapper()),
167+
default_partition_mapper=ProductMapper(IdentityMapper(), StartOfDayMapper()),
168168
),
169169
catchup=False,
170170
tags=["sales", "aggregation"],
@@ -173,7 +173,7 @@ def ingest_sales():
173173
Aggregate regional sales using ProductMapper.
174174
175175
The ProductMapper splits the composite key "region|timestamp" and applies
176-
IdentityMapper to the region segment and ToDailyMapper to the timestamp segment,
176+
IdentityMapper to the region segment and StartOfDayMapper to the timestamp segment,
177177
aligning hourly partitions to daily granularity per region.
178178
"""
179179

‎airflow-core/src/airflow/partition_mappers/temporal.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
6363
)
6464

6565

66-
class ToHourlyMapper(_BaseTemporalMapper):
66+
class StartOfHourMapper(_BaseTemporalMapper):
6767
"""Map a time-based partition key to hour."""
6868

6969
default_output_format = "%Y-%m-%dT%H"
@@ -72,7 +72,7 @@ def normalize(self, dt: datetime) -> datetime:
7272
return dt.replace(minute=0, second=0, microsecond=0)
7373

7474

75-
class ToDailyMapper(_BaseTemporalMapper):
75+
class StartOfDayMapper(_BaseTemporalMapper):
7676
"""Map a time-based partition key to day."""
7777

7878
default_output_format = "%Y-%m-%d"
@@ -81,7 +81,7 @@ def normalize(self, dt: datetime) -> datetime:
8181
return dt.replace(hour=0, minute=0, second=0, microsecond=0)
8282

8383

84-
class ToWeeklyMapper(_BaseTemporalMapper):
84+
class StartOfWeekMapper(_BaseTemporalMapper):
8585
"""Map a time-based partition key to week."""
8686

8787
default_output_format = "%Y-%m-%d (W%V)"
@@ -91,7 +91,7 @@ def normalize(self, dt: datetime) -> datetime:
9191
return start.replace(hour=0, minute=0, second=0, microsecond=0)
9292

9393

94-
class ToMonthlyMapper(_BaseTemporalMapper):
94+
class StartOfMonthMapper(_BaseTemporalMapper):
9595
"""Map a time-based partition key to month."""
9696

9797
default_output_format = "%Y-%m"
@@ -106,7 +106,7 @@ def normalize(self, dt: datetime) -> datetime:
106106
)
107107

108108

109-
class ToQuarterlyMapper(_BaseTemporalMapper):
109+
class StartOfQuarterMapper(_BaseTemporalMapper):
110110
"""Map a time-based partition key to quarter."""
111111

112112
default_output_format = "%Y-Q{quarter}"
@@ -128,7 +128,7 @@ def format(self, dt: datetime) -> str:
128128
return dt.strftime(self.output_format).format(quarter=quarter)
129129

130130

131-
class ToYearlyMapper(_BaseTemporalMapper):
131+
class StartOfYearMapper(_BaseTemporalMapper):
132132
"""Map a time-based partition key to year."""
133133

134134
default_output_format = "%Y"

‎airflow-core/src/airflow/serialization/encoders.py‎

Lines changed: 25 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -44,15 +44,15 @@
4444
MultipleCronTriggerTimetable,
4545
PartitionMapper,
4646
ProductMapper,
47-
ToDailyMapper,
48-
ToMonthlyMapper,
49-
ToQuarterlyMapper,
50-
ToWeeklyMapper,
51-
ToYearlyMapper,
47+
StartOfDayMapper,
48+
StartOfMonthMapper,
49+
StartOfQuarterMapper,
50+
StartOfWeekMapper,
51+
StartOfYearMapper,
5252
)
5353
from airflow.sdk.bases.timetable import BaseTimetable
5454
from airflow.sdk.definitions.asset import AssetRef
55-
from airflow.sdk.definitions.partition_mappers.temporal import ToHourlyMapper
55+
from airflow.sdk.definitions.partition_mappers.temporal import StartOfHourMapper
5656
from airflow.sdk.definitions.timetables.assets import (
5757
AssetTriggeredTimetable,
5858
PartitionedAssetTimetable,
@@ -393,16 +393,16 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
393393
}
394394

395395
BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
396+
AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
396397
ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
397398
IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
398-
ToHourlyMapper: "airflow.partition_mappers.temporal.ToHourlyMapper",
399-
ToDailyMapper: "airflow.partition_mappers.temporal.ToDailyMapper",
400-
ToWeeklyMapper: "airflow.partition_mappers.temporal.ToWeeklyMapper",
401-
ToMonthlyMapper: "airflow.partition_mappers.temporal.ToMonthlyMapper",
402-
ToQuarterlyMapper: "airflow.partition_mappers.temporal.ToQuarterlyMapper",
403-
ToYearlyMapper: "airflow.partition_mappers.temporal.ToYearlyMapper",
404399
ProductMapper: "airflow.partition_mappers.product.ProductMapper",
405-
AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
400+
StartOfDayMapper: "airflow.partition_mappers.temporal.StartOfDayMapper",
401+
StartOfHourMapper: "airflow.partition_mappers.temporal.StartOfHourMapper",
402+
StartOfMonthMapper: "airflow.partition_mappers.temporal.StartOfMonthMapper",
403+
StartOfQuarterMapper: "airflow.partition_mappers.temporal.StartOfQuarterMapper",
404+
StartOfWeekMapper: "airflow.partition_mappers.temporal.StartOfWeekMapper",
405+
StartOfYearMapper: "airflow.partition_mappers.temporal.StartOfYearMapper",
406406
}
407407

408408
@functools.singledispatchmethod
@@ -421,20 +421,20 @@ def _(self, partition_mapper: ChainMapper) -> dict[str, Any]:
421421
def _(self, partition_mapper: IdentityMapper) -> dict[str, Any]:
422422
return {}
423423

424-
@serialize_partition_mapper.register(ToHourlyMapper)
425-
@serialize_partition_mapper.register(ToDailyMapper)
426-
@serialize_partition_mapper.register(ToWeeklyMapper)
427-
@serialize_partition_mapper.register(ToMonthlyMapper)
428-
@serialize_partition_mapper.register(ToQuarterlyMapper)
429-
@serialize_partition_mapper.register(ToYearlyMapper)
424+
@serialize_partition_mapper.register(StartOfHourMapper)
425+
@serialize_partition_mapper.register(StartOfDayMapper)
426+
@serialize_partition_mapper.register(StartOfWeekMapper)
427+
@serialize_partition_mapper.register(StartOfMonthMapper)
428+
@serialize_partition_mapper.register(StartOfQuarterMapper)
429+
@serialize_partition_mapper.register(StartOfYearMapper)
430430
def _(
431431
self,
432-
partition_mapper: ToHourlyMapper
433-
| ToDailyMapper
434-
| ToWeeklyMapper
435-
| ToMonthlyMapper
436-
| ToQuarterlyMapper
437-
| ToYearlyMapper,
432+
partition_mapper: StartOfHourMapper
433+
| StartOfDayMapper
434+
| StartOfWeekMapper
435+
| StartOfMonthMapper
436+
| StartOfQuarterMapper
437+
| StartOfYearMapper,
438438
) -> dict[str, Any]:
439439
return {
440440
"input_format": partition_mapper.input_format,

‎airflow-core/tests/unit/jobs/test_scheduler_job.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@
9292
AssetWatcher,
9393
CronPartitionTimetable,
9494
IdentityMapper,
95-
ToHourlyMapper,
95+
StartOfHourMapper,
9696
task,
9797
)
9898
from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
@@ -8982,7 +8982,7 @@ def test_partitioned_dag_run_with_invalid_mapping(dag_maker: DagMaker, session:
89828982
dag_id="asset-event-consumer",
89838983
schedule=PartitionedAssetTimetable(
89848984
assets=asset_1,
8985-
default_partition_mapper=ToHourlyMapper(),
8985+
default_partition_mapper=StartOfHourMapper(),
89868986
),
89878987
session=session,
89888988
):

‎airflow-core/tests/unit/partition_mappers/test_chain.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
from airflow.partition_mappers.base import PartitionMapper
2323
from airflow.partition_mappers.chain import ChainMapper
2424
from airflow.partition_mappers.identity import IdentityMapper
25-
from airflow.partition_mappers.temporal import ToDailyMapper, ToHourlyMapper
25+
from airflow.partition_mappers.temporal import StartOfDayMapper, StartOfHourMapper
2626

2727

2828
class _InvalidReturnMapper(PartitionMapper):
@@ -37,7 +37,7 @@ def to_downstream(self, key: str) -> list[None]: # type: ignore[override]
3737

3838
class TestChainMapper:
3939
def test_to_downstream(self):
40-
sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
40+
sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
4141
assert sm.to_downstream("2024-01-15T10:30:00") == "2024-01-15"
4242

4343
def test_to_downstream_invalid_non_iterable_return(self):
@@ -53,17 +53,17 @@ def test_to_downstream_invalid_iterable_contents(self):
5353
def test_serialize(self):
5454
from airflow.serialization.encoders import encode_partition_mapper
5555

56-
sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
56+
sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
5757
result = sm.serialize()
5858
assert result == {
5959
"mappers": [
60-
encode_partition_mapper(ToHourlyMapper()),
61-
encode_partition_mapper(ToDailyMapper(input_format="%Y-%m-%dT%H")),
60+
encode_partition_mapper(StartOfHourMapper()),
61+
encode_partition_mapper(StartOfDayMapper(input_format="%Y-%m-%dT%H")),
6262
],
6363
}
6464

6565
def test_deserialize(self):
66-
sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
66+
sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
6767
serialized = sm.serialize()
6868
restored = ChainMapper.deserialize(serialized)
6969
assert isinstance(restored, ChainMapper)

0 commit comments

Comments
 (0)