uranusjr commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4508270639
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -110,28 +118,118 @@ def _bundle_item_exc(msg):
)
-class _BundleConfigSnapshot(NamedTuple):
- """The configured Dag bundles, as names only and as full configs."""
+@dataclass
+class _BundleConfigSnapshot:
+ """The last accepted bundle configuration in this process."""
- configs: tuple[_ExternalBundleConfig, ...]
+ configs: dict[str, _ExternalBundleConfig]
names: frozenset[str]
+ team_names: dict[str, str | None]
+ last_checked: float = field(default_factory=lambda: time.monotonic())
+
+
+def _get_bundle_config_reload_option() -> tuple[str, str] | None:
+ section, key = "dag_processor", "dag_bundle_config_list"
+ env_key = "AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST"
+ if env_key in os.environ:
+ return None
+ for suffix in ("cmd", "secret"):
+ env_option = f"{env_key}_{suffix.upper()}"
+ if env_option in os.environ:
+ return suffix, os.environ[env_option]
+ if ConfigParser.has_option(conf, section, key):
+ return None
+ for suffix in ("cmd", "secret"):
+ option = f"{key}_{suffix}"
+ if ConfigParser.has_option(conf, section, option):
+ return suffix, ConfigParser.get(conf, section, option)
+ return None
[email protected]
-def _load_bundle_config_snapshot() -> _BundleConfigSnapshot:
- """Read and validate the configured Dag bundles, without importing their
classes."""
- bundle_config_list = _read_bundle_config_list()
- if not bundle_config_list:
- return _BundleConfigSnapshot(configs=(), names=frozenset())
- if conf.getboolean("core", "LOAD_EXAMPLES"):
+def _create_bundle_config_snapshot(bundle_config_list:
list[_ExternalBundleConfig]) -> _BundleConfigSnapshot:
+ team_names = {cfg.name: cfg.team_name for cfg in bundle_config_list}
+ if bundle_config_list and conf.getboolean("core", "LOAD_EXAMPLES"):
_add_example_dag_bundle(bundle_config_list)
_add_provider_example_dags_to_bundle(bundle_config_list)
return _BundleConfigSnapshot(
- configs=tuple(bundle_config_list),
+ configs={cfg.name: cfg for cfg in bundle_config_list},
names=frozenset(cfg.name for cfg in bundle_config_list),
+ team_names=team_names,
)
+def is_bundle_config_reload_enabled() -> bool:
+ """Return whether bundle configuration comes from a command or secrets
backend."""
+ return _get_bundle_config_reload_option() is not None
+
+
[email protected]
+def _get_cached_bundle_config_snapshot(*, metadata_only: bool = False) ->
_BundleConfigSnapshot:
+ """Return the cached bundle snapshot, reading configuration on first
use."""
+ if metadata_only:
+ return replace(_get_cached_bundle_config_snapshot())
+ return _create_bundle_config_snapshot(_read_bundle_config_list())
Review Comment:
The cached return value is then mutated in place. A second cache entry
(`metadata_only=True`) is a `replace()` copy that then drifts on its own, with
its own `last_checked` and a second command run.
Replace this with a set small private classes, e.g.
`_StaticBundleConfigSource` and `_ReloadingBundleConfigSource` behind an
interface (`refresh()` + `current()` maybe), and choose between them once per
process. It then owns the snapshot, the throttle and the miss cache. Every
`is_bundle_config_reload_enabled()` branch becomes polymorphic dispatch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
