style: remove core partition mapper from typing

Author: Lee-W

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

@@ -8748,7 +8748,12 @@ def test_partitioned_dag_run_with_customized_mapper(
             dag_id="asset-event-consumer",
             schedule=PartitionedAssetTimetable(
                 assets=Asset(name="asset-1"),
-                default_partition_mapper=Key1Mapper(),
+                # Most users should use the partition mapper provided by the task-SDK.
+                # Advanced users can import from core and register their own partition mapper
+                # via an Airflow plugin.
+                # We intentionally exclude core mappers from the public typing
+                # so standard users don't accidentally rely on internal implementations.
+                default_partition_mapper=Key1Mapper(),  # type: ignore[arg-type]
             ),
             session=session,
         ):
@@ -8896,9 +8901,14 @@ def test_consumer_dag_listen_to_two_partitioned_asset_with_key_1_mapper(
             dag_id="asset-event-consumer",
             schedule=PartitionedAssetTimetable(
                 assets=(Asset(name="asset-1") & Asset(name="asset-2")),
+                # Most users should use the partition mapper provided by the task-SDK.
+                # Advanced users can import from core and register their own partition mapper
+                # via an Airflow plugin.
+                # We intentionally exclude core mappers from the public typing
+                # so standard users don't accidentally rely on internal implementations.
                 partition_mapper_mapping={
-                    Asset(name="asset-1"): Key1Mapper(),
-                    Asset(name="asset-2"): Key1Mapper(),
+                    Asset(name="asset-1"): Key1Mapper(),  # type: ignore[dict-item]
+                    Asset(name="asset-2"): Key1Mapper(),  # type: ignore[dict-item]
                 },
             ),
             session=session,

task-sdk/src/airflow/sdk/definitions/timetables/assets.py

@@ -28,7 +28,6 @@
 if TYPE_CHECKING:
     from collections.abc import Collection
 
-    from airflow.partition_mapper.base import PartitionMapper as CorePartitionMapper
     from airflow.sdk import Asset
     from airflow.sdk.definitions.partition_mapper.base import PartitionMapper
 
@@ -51,10 +50,8 @@ class PartitionedAssetTimetable(AssetTriggeredTimetable):
     """Asset-driven timetable that listens for partitioned assets."""
 
     asset_condition: BaseAsset = attrs.field(alias="assets")
-    partition_mapper_mapping: dict[BaseAsset, PartitionMapper | CorePartitionMapper] = attrs.field(
-        factory=dict
-    )
-    default_partition_mapper: PartitionMapper | CorePartitionMapper = IdentityMapper()
+    partition_mapper_mapping: dict[BaseAsset, PartitionMapper] = attrs.field(factory=dict)
+    default_partition_mapper: PartitionMapper = IdentityMapper()
 
 
 def _coerce_assets(o: Collection[Asset] | BaseAsset) -> BaseAsset: