samraj2k commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4517227843
##########
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:
Thanks for the review. I agree with your idea, and initially didnt do this
to not introduce any public class, but your segregation is mostly through
private class.
As per the change, there is now a small private source with two
implementations (reloading, static), along with: cache of missing names is per
process now, not per manager.
Though, I have not yet removed the `is_bundle_config_reload_enabled()`
checks at the call sites, or the fallbacks for when no context is passed to
retain existing behaviour (becoz: sending the context in static mode also would
change the parsing path for every deployment including ones that never use
reload), but this can be taken in a followup.
Nostly, the changes were for refactoring, but I have retested, 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]
