This is an automated email from the ASF dual-hosted git repository.
uranusjr 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 8b5306690ac Fix 'lineage' export in airflow.sdk (#75744)
8b5306690ac is described below
commit 8b5306690acb58d0584376272dfa2159aecea0dd
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Tue Oct 27 22:15:19 2026 +0800
Fix 'lineage' export in airflow.sdk (#75744)
---
.../tests/unit/partition_mappers/test_init.py | 4 +-
task-sdk/src/airflow/sdk/__init__.py | 227 +++++++++++----------
task-sdk/tests/task_sdk/docs/test_public_api.py | 26 ++-
3 files changed, 136 insertions(+), 121 deletions(-)
diff --git a/airflow-core/tests/unit/partition_mappers/test_init.py
b/airflow-core/tests/unit/partition_mappers/test_init.py
index e7579ec6eec..b7d7d7fc301 100644
--- a/airflow-core/tests/unit/partition_mappers/test_init.py
+++ b/airflow-core/tests/unit/partition_mappers/test_init.py
@@ -24,7 +24,7 @@ def test_package_root_exports_match_sdk():
"""Every partition-mapper symbol exported from airflow.sdk must also be
exported from the core package root."""
sdk_names = {
name
- for name, module in airflow.sdk.__lazy_imports.items()
- if module.startswith(".definitions.partition_mappers")
+ for name, (module_path, _) in airflow.sdk.__lazy_imports.items()
+ if module_path.startswith(".definitions.partition_mappers")
}
assert sdk_names == set(airflow.partition_mappers.__all__)
diff --git a/task-sdk/src/airflow/sdk/__init__.py
b/task-sdk/src/airflow/sdk/__init__.py
index 3a2a2fb4a0a..46418ceeaf1 100644
--- a/task-sdk/src/airflow/sdk/__init__.py
+++ b/task-sdk/src/airflow/sdk/__init__.py
@@ -126,6 +126,7 @@ __all__ = [
__version__ = "1.4.0"
if TYPE_CHECKING:
+ from airflow.sdk import lineage
from airflow.sdk.api.datamodels._generated import DagRunState,
TaskInstanceState, TriggerRule, WeightRule
from airflow.sdk.bases.branch import BaseBranchOperator, BranchMixIn
from airflow.sdk.bases.hook import BaseHook
@@ -237,120 +238,124 @@ if TYPE_CHECKING:
conf: AirflowSDKConfigParser
-__lazy_imports: dict[str, str] = {
- "AllowedKeyMapper": ".definitions.partition_mappers.allowed_key",
- "Asset": ".definitions.asset",
- "AssetAccessControl": ".definitions.asset",
- "AssetAlias": ".definitions.asset",
- "AssetAll": ".definitions.asset",
- "AssetAndTimeSchedule": ".definitions.timetables.assets",
- "AssetAny": ".definitions.asset",
- "AssetOrTimeSchedule": ".definitions.timetables.assets",
- "AssetTriggeredTimetable": ".definitions.timetables.assets",
- "AssetWatcher": ".definitions.asset",
- "AsyncCallback": ".definitions.callback",
- "BaseAsyncOperator": ".bases.operator",
- "BaseBranchOperator": ".bases.branch",
- "BaseDeadlineReference": ".definitions.deadline",
- "BaseHook": ".bases.hook",
- "BaseNotifier": ".bases.notifier",
- "BaseOperator": ".bases.operator",
- "BaseOperatorLink": ".bases.operatorlink",
- "BaseSensorOperator": ".bases.sensor",
- "BaseXCom": ".bases.xcom",
- "BranchMixIn": ".bases.branch",
- "ChainMapper": ".definitions.partition_mappers.chain",
- "ChainRetryPolicy": ".definitions.retry_policy",
- "Connection": ".definitions.connection",
- "Context": ".definitions.context",
- "CronDataIntervalTimetable": ".definitions.timetables.interval",
- "CronTriggerTimetable": ".definitions.timetables.trigger",
- "CronPartitionTimetable": ".definitions.timetables.trigger",
- "DAG": ".definitions.dag",
- "DagRunState": ".api.datamodels._generated",
- "DayWindow": ".definitions.partition_mappers.window",
- "DeadlineAlert": ".definitions.deadline",
- "DeadlineReference": ".definitions.deadline",
- "DeltaDataIntervalTimetable": ".definitions.timetables.interval",
- "DeltaTriggerTimetable": ".definitions.timetables.trigger",
- "EdgeModifier": ".definitions.edges",
- "EventsTimetable": ".definitions.timetables.events",
- "ExceptionRetryPolicy": ".definitions.retry_policy",
- "FanOutMapper": ".definitions.partition_mappers.temporal",
- "FixedKeyMapper": ".definitions.partition_mappers.fixed_key",
- "HourWindow": ".definitions.partition_mappers.window",
- "IdentityMapper": ".definitions.partition_mappers.identity",
- "Label": ".definitions.edges",
- "LoopDecision": ".definitions.loop",
- "Metadata": ".definitions.asset.metadata",
- "MinimumCount": ".definitions.partition_mappers.wait_policy",
- "MonthWindow": ".definitions.partition_mappers.window",
- "MultipleCronTriggerTimetable": ".definitions.timetables.trigger",
- "ObjectStoragePath": ".io.path",
- "Param": ".definitions.param",
- "ParamsDict": ".definitions.param",
- "PartitionedAtRuntime": ".definitions.timetables.assets",
- "PartitionedAssetTimetable": ".definitions.timetables.assets",
- "PartitionMapper": ".definitions.partition_mappers.base",
- "PokeReturnValue": ".bases.sensor",
- "ProductMapper": ".definitions.partition_mappers.product",
- "QuarterWindow": ".definitions.partition_mappers.window",
- "ResumableJobMixin": ".bases.resumablejobmixin",
- "RetryAction": ".definitions.retry_policy",
- "RetryDecision": ".definitions.retry_policy",
- "RetryPolicy": ".definitions.retry_policy",
- "RetryRule": ".definitions.retry_policy",
- "RollupMapper": ".definitions.partition_mappers.base",
- "SecretCache": ".execution_time.cache",
- "SegmentWindow": ".definitions.partition_mappers.window",
- "SkipMixin": ".bases.skipmixin",
- "SyncCallback": ".definitions.callback",
- "StartOfDayMapper": ".definitions.partition_mappers.temporal",
- "StartOfHourMapper": ".definitions.partition_mappers.temporal",
- "StartOfMonthMapper": ".definitions.partition_mappers.temporal",
- "StartOfQuarterMapper": ".definitions.partition_mappers.temporal",
- "StartOfWeekMapper": ".definitions.partition_mappers.temporal",
- "StartOfYearMapper": ".definitions.partition_mappers.temporal",
- "TaskGroup": ".definitions.taskgroup",
- "TaskInstance": ".types",
- "TaskInstanceState": ".api.datamodels._generated",
- "TriggerRule": ".api.datamodels._generated",
- "Variable": ".definitions.variable",
- "WaitForAll": ".definitions.partition_mappers.wait_policy",
- "WeekWindow": ".definitions.partition_mappers.window",
- "WeightRule": ".api.datamodels._generated",
- "Window": ".definitions.partition_mappers.window",
- "XComArg": ".definitions.xcom_arg",
- "YearWindow": ".definitions.partition_mappers.window",
- "asset": ".definitions.asset.decorators",
- "chain": ".bases.operator",
- "chain_linear": ".bases.operator",
- "conf": ".configuration",
- "cross_downstream": ".bases.operator",
- "dag": ".definitions.dag",
- "deadline_reference": ".definitions.deadline",
- "NEVER_EXPIRE": ".execution_time.context",
- "get_current_context": ".definitions.context",
- "get_parsing_context": ".definitions.context",
- "literal": ".definitions.template",
- "lineage": ".lineage",
- "macros": ".execution_time",
- "result": ".definitions.decorators",
- "setup": ".definitions.decorators",
- "task": ".definitions.decorators",
- "task_group": ".definitions.decorators",
- "teardown": ".definitions.decorators",
+# Things to lazy import in form {local_name: (target_module, target_name)}.
+# An empty target_name exposes the target module itself.
+__lazy_imports: dict[str, tuple[str, str]] = {
+ "AllowedKeyMapper": (".definitions.partition_mappers.allowed_key",
"AllowedKeyMapper"),
+ "Asset": (".definitions.asset", "Asset"),
+ "AssetAccessControl": (".definitions.asset", "AssetAccessControl"),
+ "AssetAlias": (".definitions.asset", "AssetAlias"),
+ "AssetAll": (".definitions.asset", "AssetAll"),
+ "AssetAndTimeSchedule": (".definitions.timetables.assets",
"AssetAndTimeSchedule"),
+ "AssetAny": (".definitions.asset", "AssetAny"),
+ "AssetOrTimeSchedule": (".definitions.timetables.assets",
"AssetOrTimeSchedule"),
+ "AssetTriggeredTimetable": (".definitions.timetables.assets",
"AssetTriggeredTimetable"),
+ "AssetWatcher": (".definitions.asset", "AssetWatcher"),
+ "AsyncCallback": (".definitions.callback", "AsyncCallback"),
+ "BaseAsyncOperator": (".bases.operator", "BaseAsyncOperator"),
+ "BaseBranchOperator": (".bases.branch", "BaseBranchOperator"),
+ "BaseDeadlineReference": (".definitions.deadline",
"BaseDeadlineReference"),
+ "BaseHook": (".bases.hook", "BaseHook"),
+ "BaseNotifier": (".bases.notifier", "BaseNotifier"),
+ "BaseOperator": (".bases.operator", "BaseOperator"),
+ "BaseOperatorLink": (".bases.operatorlink", "BaseOperatorLink"),
+ "BaseSensorOperator": (".bases.sensor", "BaseSensorOperator"),
+ "BaseXCom": (".bases.xcom", "BaseXCom"),
+ "BranchMixIn": (".bases.branch", "BranchMixIn"),
+ "ChainMapper": (".definitions.partition_mappers.chain", "ChainMapper"),
+ "ChainRetryPolicy": (".definitions.retry_policy", "ChainRetryPolicy"),
+ "Connection": (".definitions.connection", "Connection"),
+ "Context": (".definitions.context", "Context"),
+ "CronDataIntervalTimetable": (".definitions.timetables.interval",
"CronDataIntervalTimetable"),
+ "CronTriggerTimetable": (".definitions.timetables.trigger",
"CronTriggerTimetable"),
+ "CronPartitionTimetable": (".definitions.timetables.trigger",
"CronPartitionTimetable"),
+ "DAG": (".definitions.dag", "DAG"),
+ "DagRunState": (".api.datamodels._generated", "DagRunState"),
+ "DayWindow": (".definitions.partition_mappers.window", "DayWindow"),
+ "DeadlineAlert": (".definitions.deadline", "DeadlineAlert"),
+ "DeadlineReference": (".definitions.deadline", "DeadlineReference"),
+ "DeltaDataIntervalTimetable": (".definitions.timetables.interval",
"DeltaDataIntervalTimetable"),
+ "DeltaTriggerTimetable": (".definitions.timetables.trigger",
"DeltaTriggerTimetable"),
+ "EdgeModifier": (".definitions.edges", "EdgeModifier"),
+ "EventsTimetable": (".definitions.timetables.events", "EventsTimetable"),
+ "ExceptionRetryPolicy": (".definitions.retry_policy",
"ExceptionRetryPolicy"),
+ "FanOutMapper": (".definitions.partition_mappers.temporal",
"FanOutMapper"),
+ "FixedKeyMapper": (".definitions.partition_mappers.fixed_key",
"FixedKeyMapper"),
+ "HourWindow": (".definitions.partition_mappers.window", "HourWindow"),
+ "IdentityMapper": (".definitions.partition_mappers.identity",
"IdentityMapper"),
+ "Label": (".definitions.edges", "Label"),
+ "LoopDecision": (".definitions.loop", "LoopDecision"),
+ "Metadata": (".definitions.asset.metadata", "Metadata"),
+ "MinimumCount": (".definitions.partition_mappers.wait_policy",
"MinimumCount"),
+ "MonthWindow": (".definitions.partition_mappers.window", "MonthWindow"),
+ "MultipleCronTriggerTimetable": (".definitions.timetables.trigger",
"MultipleCronTriggerTimetable"),
+ "ObjectStoragePath": (".io.path", "ObjectStoragePath"),
+ "Param": (".definitions.param", "Param"),
+ "ParamsDict": (".definitions.param", "ParamsDict"),
+ "PartitionedAtRuntime": (".definitions.timetables.assets",
"PartitionedAtRuntime"),
+ "PartitionedAssetTimetable": (".definitions.timetables.assets",
"PartitionedAssetTimetable"),
+ "PartitionMapper": (".definitions.partition_mappers.base",
"PartitionMapper"),
+ "PokeReturnValue": (".bases.sensor", "PokeReturnValue"),
+ "ProductMapper": (".definitions.partition_mappers.product",
"ProductMapper"),
+ "QuarterWindow": (".definitions.partition_mappers.window",
"QuarterWindow"),
+ "ResumableJobMixin": (".bases.resumablejobmixin", "ResumableJobMixin"),
+ "RetryAction": (".definitions.retry_policy", "RetryAction"),
+ "RetryDecision": (".definitions.retry_policy", "RetryDecision"),
+ "RetryPolicy": (".definitions.retry_policy", "RetryPolicy"),
+ "RetryRule": (".definitions.retry_policy", "RetryRule"),
+ "RollupMapper": (".definitions.partition_mappers.base", "RollupMapper"),
+ "SecretCache": (".execution_time.cache", "SecretCache"),
+ "SegmentWindow": (".definitions.partition_mappers.window",
"SegmentWindow"),
+ "SkipMixin": (".bases.skipmixin", "SkipMixin"),
+ "SyncCallback": (".definitions.callback", "SyncCallback"),
+ "StartOfDayMapper": (".definitions.partition_mappers.temporal",
"StartOfDayMapper"),
+ "StartOfHourMapper": (".definitions.partition_mappers.temporal",
"StartOfHourMapper"),
+ "StartOfMonthMapper": (".definitions.partition_mappers.temporal",
"StartOfMonthMapper"),
+ "StartOfQuarterMapper": (".definitions.partition_mappers.temporal",
"StartOfQuarterMapper"),
+ "StartOfWeekMapper": (".definitions.partition_mappers.temporal",
"StartOfWeekMapper"),
+ "StartOfYearMapper": (".definitions.partition_mappers.temporal",
"StartOfYearMapper"),
+ "TaskGroup": (".definitions.taskgroup", "TaskGroup"),
+ "TaskInstance": (".types", "TaskInstance"),
+ "TaskInstanceState": (".api.datamodels._generated", "TaskInstanceState"),
+ "TriggerRule": (".api.datamodels._generated", "TriggerRule"),
+ "Variable": (".definitions.variable", "Variable"),
+ "WaitForAll": (".definitions.partition_mappers.wait_policy", "WaitForAll"),
+ "WeekWindow": (".definitions.partition_mappers.window", "WeekWindow"),
+ "WeightRule": (".api.datamodels._generated", "WeightRule"),
+ "Window": (".definitions.partition_mappers.window", "Window"),
+ "XComArg": (".definitions.xcom_arg", "XComArg"),
+ "YearWindow": (".definitions.partition_mappers.window", "YearWindow"),
+ "asset": (".definitions.asset.decorators", "asset"),
+ "chain": (".bases.operator", "chain"),
+ "chain_linear": (".bases.operator", "chain_linear"),
+ "conf": (".configuration", "conf"),
+ "cross_downstream": (".bases.operator", "cross_downstream"),
+ "dag": (".definitions.dag", "dag"),
+ "deadline_reference": (".definitions.deadline", "deadline_reference"),
+ "NEVER_EXPIRE": (".execution_time.context", "NEVER_EXPIRE"),
+ "get_current_context": (".definitions.context", "get_current_context"),
+ "get_parsing_context": (".definitions.context", "get_parsing_context"),
+ "literal": (".definitions.template", "literal"),
+ "lineage": (".lineage", ""),
+ "macros": (".execution_time.macros", ""),
+ "result": (".definitions.decorators", "result"),
+ "setup": (".definitions.decorators", "setup"),
+ "task": (".definitions.decorators", "task"),
+ "task_group": (".definitions.decorators", "task_group"),
+ "teardown": (".definitions.decorators", "teardown"),
}
def __getattr__(name: str):
- if module_path := __lazy_imports.get(name):
- import importlib
+ module_path, attr_name = __lazy_imports.get(name, ("", ""))
+ if not module_path:
+ raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
- mod = importlib.import_module(module_path, __name__)
- val = getattr(mod, name)
+ import importlib
- # Store for next time
- globals()[name] = val
- return val
- raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+ mod = importlib.import_module(module_path, __name__)
+ val = getattr(mod, attr_name) if attr_name else mod
+
+ # Store for next time
+ globals()[name] = val
+ return val
diff --git a/task-sdk/tests/task_sdk/docs/test_public_api.py
b/task-sdk/tests/task_sdk/docs/test_public_api.py
index eea7740cc4b..0b00b32bc28 100644
--- a/task-sdk/tests/task_sdk/docs/test_public_api.py
+++ b/task-sdk/tests/task_sdk/docs/test_public_api.py
@@ -17,19 +17,29 @@
from __future__ import annotations
import importlib
+import subprocess
+import sys
+import textwrap
def test_airflow_sdk_exports_exist():
"""
- Ensure that all names declared in airflow.sdk.__all__ are present as
attributes on the module.
+ Ensure every name in airflow.sdk.__all__ resolves through the lazy
``__getattr__``.
+
+ This runs in a subprocess because in a shared test process, a submodule
export
+ (e.g. ``lineage``) may already be bound on the package as an import side
effect,
+ which would mask a broken lazy-import entry.
"""
- sdk = importlib.import_module("airflow.sdk")
- # Provide literal attribute for testing since it's declared in __all__
- template_mod = importlib.import_module("airflow.sdk.definitions.template")
- setattr(sdk, "literal", getattr(template_mod, "literal"))
- public_names = getattr(sdk, "__all__", [])
- missing = [name for name in public_names if not hasattr(sdk, name)]
- assert not missing, f"Missing exports in airflow.sdk: {missing}"
+ code = textwrap.dedent(
+ """
+ import airflow.sdk as sdk
+
+ missing = [name for name in sdk.__all__ if not hasattr(sdk, name)]
+ assert not missing, f"Missing exports in airflow.sdk: {missing}"
+ """
+ )
+ result = subprocess.run([sys.executable, "-c", code], check=False,
capture_output=True, text=True)
+ assert result.returncode == 0, result.stderr
def test_airflow_sdk_no_unexpected_exports():
