samraj2k commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4441559100
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -108,28 +115,116 @@ 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())
+
+
+def _refresh_bundle_config_snapshot(
+ *,
+ force: bool = False,
+ metadata_only: bool = False,
+ snapshot: _BundleConfigSnapshot | None = None,
+) -> _BundleConfigSnapshot:
+ cached = (
+ _get_cached_bundle_config_snapshot(metadata_only=True)
+ if metadata_only
+ else _get_cached_bundle_config_snapshot()
+ )
+ if snapshot is None:
+ snapshot = cached
+ if not is_bundle_config_reload_enabled():
+ return snapshot
+ now = time.monotonic()
+ try:
+ if not force and now - cached.last_checked < conf.getint(
+ "dag_processor", "bundle_refresh_check_interval"
+ ):
+ if snapshot.configs is cached.configs:
+ return snapshot
+ config_list = [cached.configs[name] for name in cached.team_names]
+ else:
+ cached.last_checked = now
+ config_list = _read_bundle_config_list()
+ if {config.name: config for config in config_list} == {
+ name: snapshot.configs[name] for name in snapshot.team_names
+ }:
+ current = snapshot
+ else:
+ for config in config_list:
+ existing = snapshot.configs.get(config.name)
+ if existing is not None:
+ if existing != config:
+ raise _bundle_item_exc(
+ f"Use a new bundle name when changing
configuration for '{config.name}'"
Review Comment:
Thanks @potiuk, this behaviour for holding the list if some bundle
definition is changed is intentional. From your feedback, I have enhanced the
comment and documentation, thanks
--
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]
