From 7d323f78ef79ea734cd53ce0870109dfaec2e5e7 Mon Sep 17 00:00:00 2001 From: David Blain Date: Mon, 13 Jul 2026 12:45:01 +0200 Subject: [PATCH 1/7] refactor: Centralise structlog initialisation in BaseDagBundle --- .../airflow/dag_processing/bundles/base.py | 15 ++++++++++++ .../providers/amazon/aws/bundles/s3.py | 18 ++++++-------- .../src/airflow/providers/git/bundles/git.py | 24 ++++++++----------- .../providers/google/cloud/bundles/gcs.py | 17 ++++++------- 4 files changed, 39 insertions(+), 35 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py b/airflow-core/src/airflow/dag_processing/bundles/base.py index 344a3349fecab..13d61f4871dc5 100644 --- a/airflow-core/src/airflow/dag_processing/bundles/base.py +++ b/airflow-core/src/airflow/dag_processing/bundles/base.py @@ -33,6 +33,7 @@ from typing import TYPE_CHECKING, Any import pendulum +import structlog from pendulum.parsing import ParserError from airflow.configuration import conf @@ -309,6 +310,7 @@ def __init__( version_data: dict[str, Any] | None = None, view_url_template: str | None = None, ) -> None: + self.__log = None self.name = name self.version = version self.version_data = version_data @@ -350,6 +352,19 @@ def initialize(self) -> None: bundle_path, ) + def _log_context(self) -> dict: + """Return extra structlog context fields for this bundle. Override in subclasses.""" + return {} + + @property + def _log(self): + """Lazy structlog logger bound with common bundle context plus subclass-specific fields.""" + if self.__log is None: + self.__log = structlog.get_logger(type(self).__module__).bind( + bundle_name=self.name, version=self.version, **self._log_context() + ) + return self.__log + @property @abstractmethod def path(self) -> Path: diff --git a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py index 65bacb4b38811..8d6baceacf9b1 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py @@ -19,8 +19,6 @@ import os from pathlib import Path -import structlog - from airflow.dag_processing.bundles.base import BaseDagBundle from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook from airflow.providers.amazon.aws.hooks.s3 import S3Hook @@ -55,17 +53,15 @@ def __init__( self.prefix = prefix # Local path where S3 Dags are downloaded self.s3_dags_dir: Path = self.base_dir - - log = structlog.get_logger(__name__) - self._log = log.bind( - bundle_name=self.name, - version=self.version, - bucket_name=self.bucket_name, - prefix=self.prefix, - aws_conn_id=self.aws_conn_id, - ) self._s3_hook: S3Hook | None = None + def _log_context(self) -> dict: + return { + "bucket_name": self.bucket_name, + "prefix": self.prefix, + "aws_conn_id": self.aws_conn_id, + } + def _initialize(self): with self.lock(): if not self.s3_dags_dir.exists(): diff --git a/providers/git/src/airflow/providers/git/bundles/git.py b/providers/git/src/airflow/providers/git/bundles/git.py index 275aa6b3f634b..af48feb45b372 100644 --- a/providers/git/src/airflow/providers/git/bundles/git.py +++ b/providers/git/src/airflow/providers/git/bundles/git.py @@ -22,7 +22,6 @@ from pathlib import Path from urllib.parse import urlparse -import structlog from git import Repo from git.exc import BadName, GitCommandError, InvalidGitRepositoryError, NoSuchPathError from tenacity import retry, retry_if_exception_type, stop_after_attempt @@ -35,8 +34,6 @@ if AIRFLOW_V_3_3_PLUS: from airflow.dag_processing.bundles.base import BundleVersion -log = structlog.get_logger(__name__) - class GitDagBundle(BaseDagBundle): """ @@ -95,17 +92,6 @@ def __init__( else: self.prune_dotgit_folder = prune_dotgit_folder - self._log = log.bind( - bundle_name=self.name, - version=self.version, - bare_repo_path=self.bare_repo_path, - repo_path=self.repo_path, - versions_path=self.versions_dir, - git_conn_id=self.git_conn_id, - submodules=self.submodules, - sparse_dirs=self.sparse_dirs, - ) - self._log.debug("bundle configured") self.hook: GitHook | None = None try: @@ -150,6 +136,16 @@ def _local_repo_has_version(self) -> bool: if repo is not None: repo.close() + def _log_context(self) -> dict: + return { + "bare_repo_path": self.bare_repo_path, + "repo_path": self.repo_path, + "versions_path": self.versions_dir, + "git_conn_id": self.git_conn_id, + "submodules": self.submodules, + "sparse_dirs": self.sparse_dirs, + } + def _initialize(self): with self.lock(): # Avoids re-cloning on every task run when: diff --git a/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py b/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py index e9806276a823a..27c0e24f2a126 100644 --- a/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py @@ -19,7 +19,6 @@ import os from pathlib import Path -import structlog from google.api_core.exceptions import NotFound from airflow.dag_processing.bundles.base import BaseDagBundle @@ -56,17 +55,15 @@ def __init__( self.prefix = prefix # Local path where GCS Dags are downloaded self.gcs_dags_dir: Path = self.base_dir - - log = structlog.get_logger(__name__) - self._log = log.bind( - bundle_name=self.name, - version=self.version, - bucket_name=self.bucket_name, - prefix=self.prefix, - gcp_conn_id=self.gcp_conn_id, - ) self._gcs_hook: GCSHook | None = None + def _log_context(self) -> dict: + return { + "bucket_name": self.bucket_name, + "prefix": self.prefix, + "gcp_conn_id": self.gcp_conn_id, + } + def _initialize(self): with self.lock(): if not self.gcs_dags_dir.exists(): From 1b7170f4aaef404632ec7c1102806b37a147c40e Mon Sep 17 00:00:00 2001 From: David Blain Date: Tue, 14 Jul 2026 14:11:13 +0200 Subject: [PATCH 2/7] refactor: Implemented BaseDagBundle in common-compat so older Airflow versions can also benefit from the common log solution --- providers/amazon/pyproject.toml | 2 +- .../providers/amazon/aws/bundles/s3.py | 2 +- .../common/compat/bundles/__init__.py | 45 +++++++ .../unit/common/compat/bundles/__init__.py | 16 +++ .../common/compat/bundles/test_dag_bundle.py | 125 ++++++++++++++++++ providers/git/pyproject.toml | 2 +- .../src/airflow/providers/git/bundles/git.py | 2 +- providers/google/pyproject.toml | 2 +- .../providers/google/cloud/bundles/gcs.py | 2 +- 9 files changed, 192 insertions(+), 6 deletions(-) create mode 100644 providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py create mode 100644 providers/common/compat/tests/unit/common/compat/bundles/__init__.py create mode 100644 providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py diff --git a/providers/amazon/pyproject.toml b/providers/amazon/pyproject.toml index ef24eaea8f167..2f6c70057072b 100644 --- a/providers/amazon/pyproject.toml +++ b/providers/amazon/pyproject.toml @@ -60,7 +60,7 @@ requires-python = ">=3.10" # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build`` dependencies = [ "apache-airflow>=2.11.0", - "apache-airflow-providers-common-compat>=1.14.3", + "apache-airflow-providers-common-compat>=1.14.3", # use next version "apache-airflow-providers-common-sql>=1.32.0", "apache-airflow-providers-http", # We should update minimum version of boto3 and here regularly to avoid `pip` backtracking with the number diff --git a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py index 8d6baceacf9b1..9c92e0924c347 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py @@ -19,9 +19,9 @@ import os from pathlib import Path -from airflow.dag_processing.bundles.base import BaseDagBundle from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook from airflow.providers.amazon.aws.hooks.s3 import S3Hook +from airflow.providers.common.compat.bundles import BaseDagBundle from airflow.providers.common.compat.sdk import AirflowException diff --git a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py new file mode 100644 index 0000000000000..697d18036d5f1 --- /dev/null +++ b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py @@ -0,0 +1,45 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +import structlog + +from airflow.dag_processing.bundles.base import BaseDagBundle as _BaseDagBundle + +if hasattr(_BaseDagBundle, "_log"): + # Modern Airflow already provides BaseDagBundle._log – use it as-is. + BaseDagBundle = _BaseDagBundle +else: + # Older Airflow: inject _log via a transparent subclass so that all + # bundle implementations can use self._log without version checks. + class BaseDagBundle(_BaseDagBundle): # type: ignore[no-redef] + """Drop-in replacement for BaseDagBundle with a back-filled _log property.""" + + @property + def _log(self): + """Lazy structlog logger bound with common bundle context.""" + if not hasattr(self, "_compat_structlog"): + self._compat_structlog = structlog.get_logger(type(self).__module__).bind( + bundle_name=self.name, + version=self.version, + **self._log_context(), + ) + return self._compat_structlog + + +__all__ = ["BaseDagBundle"] diff --git a/providers/common/compat/tests/unit/common/compat/bundles/__init__.py b/providers/common/compat/tests/unit/common/compat/bundles/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/providers/common/compat/tests/unit/common/compat/bundles/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py new file mode 100644 index 0000000000000..61699f53264c8 --- /dev/null +++ b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py @@ -0,0 +1,125 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from unittest import mock + +from airflow.dag_processing.bundles.base import BaseDagBundle as CoreBaseDagBundle +from airflow.providers.common.compat.bundles import BaseDagBundle + + +class TestDagBundleCompat: + def test_base_dag_bundle_is_importable(self): + assert BaseDagBundle is not None + + def test_base_dag_bundle_is_subclass_of_core(self): + assert issubclass(BaseDagBundle, CoreBaseDagBundle) + + def test_has_log_property(self): + assert hasattr(BaseDagBundle, "_log") + + def test_log_property_on_subclass_instance(self, tmp_path): + """_log returns a bound structlog logger on concrete bundle subclasses.""" + from pathlib import Path + + from tests_common.test_utils.config import conf_vars + + class _DummyBundle(BaseDagBundle): + @property + def path(self) -> Path: + return tmp_path + + def initialize(self) -> None: + pass + + def get_current_version(self): + return None + + def refresh(self) -> None: + pass + + with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): + bundle = _DummyBundle(name="test-bundle") + + log = bundle._log + assert log is not None + # Calling _log twice returns the same (cached) logger. + assert bundle._log is log + + def test_log_is_bound_with_context(self, tmp_path): + """_log includes bundle_name and version in its bound variables.""" + from pathlib import Path + + from tests_common.test_utils.config import conf_vars + + class _DummyBundle(BaseDagBundle): + @property + def path(self) -> Path: + return tmp_path + + def initialize(self) -> None: + pass + + def get_current_version(self): + return None + + def refresh(self) -> None: + pass + + with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): + bundle = _DummyBundle(name="my-bundle", version="abc123") + + log = bundle._log + # structlog bound loggers expose their bindings via _context + ctx = log._context if hasattr(log, "_context") else {} + assert ctx.get("bundle_name") == "my-bundle" + assert ctx.get("version") == "abc123" + + def test_compat_subclass_provides_log_when_missing(self, tmp_path): + """Even when the installed BaseDagBundle has no _log, the compat class fills it in.""" + from pathlib import Path + + from tests_common.test_utils.config import conf_vars + + # Simulate an older Airflow version by temporarily hiding _log from + # the core class; the compat class must still expose it. + with mock.patch.object(CoreBaseDagBundle, "_log", None, create=True): + from importlib import reload + + import airflow.providers.common.compat.bundles as mod + + reload(mod) + CompatBase = mod.BaseDagBundle + + class _DummyBundle(CompatBase): + @property + def path(self) -> Path: + return tmp_path + + def initialize(self) -> None: + pass + + def get_current_version(self): + return None + + def refresh(self) -> None: + pass + + with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): + bundle = _DummyBundle(name="legacy-bundle") + + assert bundle._log is not None diff --git a/providers/git/pyproject.toml b/providers/git/pyproject.toml index 689f006001310..d9d52f415114b 100644 --- a/providers/git/pyproject.toml +++ b/providers/git/pyproject.toml @@ -60,7 +60,7 @@ requires-python = ">=3.10" # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build`` dependencies = [ "apache-airflow>=3.0.0", - "apache-airflow-providers-common-compat>=1.15.0", + "apache-airflow-providers-common-compat>=1.15.0", # use next version "GitPython>=3.1.44", ] diff --git a/providers/git/src/airflow/providers/git/bundles/git.py b/providers/git/src/airflow/providers/git/bundles/git.py index af48feb45b372..3f6033c813f4a 100644 --- a/providers/git/src/airflow/providers/git/bundles/git.py +++ b/providers/git/src/airflow/providers/git/bundles/git.py @@ -26,7 +26,7 @@ from git.exc import BadName, GitCommandError, InvalidGitRepositoryError, NoSuchPathError from tenacity import retry, retry_if_exception_type, stop_after_attempt -from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.providers.common.compat.bundles import BaseDagBundle from airflow.providers.common.compat.sdk import AirflowException from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_3_PLUS from airflow.providers.git.hooks.git import GitHook diff --git a/providers/google/pyproject.toml b/providers/google/pyproject.toml index d42cc3d3a2802..b49503cc80ec1 100644 --- a/providers/google/pyproject.toml +++ b/providers/google/pyproject.toml @@ -60,7 +60,7 @@ requires-python = ">=3.10" # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build`` dependencies = [ "apache-airflow>=2.11.0", - "apache-airflow-providers-common-compat>=1.13.0", + "apache-airflow-providers-common-compat>=1.13.0", # use next version "apache-airflow-providers-common-sql>=1.32.0", "asgiref>=3.5.2; python_version < '3.14'", "asgiref>=3.11.1; python_version >= '3.14'", diff --git a/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py b/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py index 27c0e24f2a126..2a99ff0dede2e 100644 --- a/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py @@ -21,7 +21,7 @@ from google.api_core.exceptions import NotFound -from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.providers.common.compat.bundles import BaseDagBundle from airflow.providers.common.compat.sdk import AirflowException from airflow.providers.google.cloud.hooks.gcs import GCSHook from airflow.providers.google.common.hooks.base_google import GoogleBaseHook From 49785d924a3f7b962f259eecb1832012141e5f11 Mon Sep 17 00:00:00 2001 From: David Blain Date: Wed, 15 Jul 2026 07:53:39 +0200 Subject: [PATCH 3/7] refactor: Skip dag bundle test if import of bundles doesn't exist --- .../tests/unit/common/compat/bundles/test_dag_bundle.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py index 61699f53264c8..0b4a84e3b0523 100644 --- a/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py +++ b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py @@ -18,6 +18,10 @@ from unittest import mock +import pytest + +pytest.importorskip("airflow.dag_processing.bundles", reason="Requires Airflow 3+") + from airflow.dag_processing.bundles.base import BaseDagBundle as CoreBaseDagBundle from airflow.providers.common.compat.bundles import BaseDagBundle From cbb18e146899433ca14f3328c6f584c2e5b00d96 Mon Sep 17 00:00:00 2001 From: David Blain Date: Wed, 15 Jul 2026 07:56:11 +0200 Subject: [PATCH 4/7] refactor: Check if BaseDagBundle has a _log property as checking existence of attribute is too weak --- .../airflow/providers/common/compat/bundles/__init__.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py index 697d18036d5f1..1ec2d9eab2076 100644 --- a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py +++ b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py @@ -17,11 +17,13 @@ from __future__ import annotations +import inspect + import structlog from airflow.dag_processing.bundles.base import BaseDagBundle as _BaseDagBundle -if hasattr(_BaseDagBundle, "_log"): +if isinstance(inspect.getattr_static(_BaseDagBundle, "_log", None), property): # Modern Airflow already provides BaseDagBundle._log – use it as-is. BaseDagBundle = _BaseDagBundle else: @@ -34,10 +36,11 @@ class BaseDagBundle(_BaseDagBundle): # type: ignore[no-redef] def _log(self): """Lazy structlog logger bound with common bundle context.""" if not hasattr(self, "_compat_structlog"): + log_context = self._log_context() if hasattr(self, "_log_context") else {} self._compat_structlog = structlog.get_logger(type(self).__module__).bind( bundle_name=self.name, version=self.version, - **self._log_context(), + **log_context, ) return self._compat_structlog From 4f3b45a90a8b3844154dac93985c8ef388fee700 Mon Sep 17 00:00:00 2001 From: David Blain Date: Thu, 16 Jul 2026 19:53:22 +0200 Subject: [PATCH 5/7] Update airflow-core/src/airflow/dag_processing/bundles/base.py Co-authored-by: Shahar Epstein <60007259+shahar1@users.noreply.github.com> --- airflow-core/src/airflow/dag_processing/bundles/base.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py b/airflow-core/src/airflow/dag_processing/bundles/base.py index 13d61f4871dc5..8e7faeaffe2a8 100644 --- a/airflow-core/src/airflow/dag_processing/bundles/base.py +++ b/airflow-core/src/airflow/dag_processing/bundles/base.py @@ -365,6 +365,11 @@ def _log(self): ) return self.__log + @_log.setter + def _log(self, value) -> None: + """Allow subclasses to bind their own logger (kept for bundles released before the lazy property).""" + self.__log = value + @property @abstractmethod def path(self) -> Path: From 1322642e8e8b40c2e72afe9b363dc4269374c3f7 Mon Sep 17 00:00:00 2001 From: David Blain Date: Thu, 16 Jul 2026 20:04:53 +0200 Subject: [PATCH 6/7] Fix _log setter, naming, types and test isolation in BaseDagBundle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Released bundle providers (git, amazon, google) assign self._log in __init__ against apache-airflow>=3.0.0 with no upper bound. Without a setter the read-only property raises AttributeError on core upgrade, breaking every installed bundle. Adding a setter keeps the lazy default while accepting direct assignment. Rename _log_context to _get_log_context so the method reads as a callable rather than an attribute (AGENTS.md naming rule). Tighten annotations: dict[str, Any] return type, -> Any on the property, and an explicit Any type on the backing __log attribute. Add tests in test_base.py that cover the lazy cache, bundle_name/version bindings, subclass _get_log_context merging, and the setter — the subclass context test is the only one that fails without the PR's change. In the compat test suite: move function-level imports to module scope, extract a legacy_compat_base fixture that reloads the module inside the patch and reloads again on teardown so sys.modules is never left in the fallback state, and replace the three identical _DummyBundle definitions with a _make_bundle_class factory. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../airflow/dag_processing/bundles/base.py | 8 +- .../unit/dag_processing/bundles/test_base.py | 53 ++++++++ .../common/compat/bundles/__init__.py | 2 +- .../common/compat/bundles/test_dag_bundle.py | 128 +++++++----------- 4 files changed, 109 insertions(+), 82 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py b/airflow-core/src/airflow/dag_processing/bundles/base.py index 8e7faeaffe2a8..c4d0288474199 100644 --- a/airflow-core/src/airflow/dag_processing/bundles/base.py +++ b/airflow-core/src/airflow/dag_processing/bundles/base.py @@ -310,7 +310,7 @@ def __init__( version_data: dict[str, Any] | None = None, view_url_template: str | None = None, ) -> None: - self.__log = None + self.__log: Any = None self.name = name self.version = version self.version_data = version_data @@ -352,16 +352,16 @@ def initialize(self) -> None: bundle_path, ) - def _log_context(self) -> dict: + def _get_log_context(self) -> dict[str, Any]: """Return extra structlog context fields for this bundle. Override in subclasses.""" return {} @property - def _log(self): + def _log(self) -> Any: """Lazy structlog logger bound with common bundle context plus subclass-specific fields.""" if self.__log is None: self.__log = structlog.get_logger(type(self).__module__).bind( - bundle_name=self.name, version=self.version, **self._log_context() + bundle_name=self.name, version=self.version, **self._get_log_context() ) return self.__log diff --git a/airflow-core/tests/unit/dag_processing/bundles/test_base.py b/airflow-core/tests/unit/dag_processing/bundles/test_base.py index f092f3e00e770..f5669f340b5d1 100644 --- a/airflow-core/tests/unit/dag_processing/bundles/test_base.py +++ b/airflow-core/tests/unit/dag_processing/bundles/test_base.py @@ -336,3 +336,56 @@ def test_version_data_defaults_to_none(): """Test that version_data defaults to None when not provided.""" bundle = BasicBundle(name="test") assert bundle.version_data is None + + +class BundleWithContext(BaseDagBundle): + @property + def path(self) -> Path: + return Path("/tmp") + + def get_current_version(self) -> str | None: ... + + def refresh(self) -> None: ... + + def _get_log_context(self) -> dict: + return {"custom_key": "custom_value"} + + +def test_log_is_lazy_and_cached(): + """_log is only created on first access and the same object is returned on repeated access.""" + bundle = BasicBundle(name="lazy-test") + assert bundle._BaseDagBundle__log is None + first = bundle._log + assert first is not None + assert bundle._log is first + + +def test_log_includes_bundle_name_and_version(): + """_log is bound with bundle_name and version.""" + import structlog.testing + + bundle = BasicBundle(name="ctx-bundle", version="v1.2.3") + with structlog.testing.capture_logs() as cap: + bundle._log.info("test event") + assert cap[0]["bundle_name"] == "ctx-bundle" + assert cap[0]["version"] == "v1.2.3" + + +def test_log_includes_subclass_context(): + """Keys returned by _get_log_context() are merged into the bound logger.""" + import structlog.testing + + bundle = BundleWithContext(name="ctx-bundle", version="v2") + with structlog.testing.capture_logs() as cap: + bundle._log.info("test event") + assert cap[0]["custom_key"] == "custom_value" + + +def test_log_setter_allows_direct_assignment(): + """Assigning self._log directly (as legacy providers do) must not raise AttributeError.""" + import structlog + + bundle = BasicBundle(name="setter-test") + custom_logger = structlog.get_logger("custom").bind(bundle_name="setter-test") + bundle._log = custom_logger + assert bundle._log is custom_logger diff --git a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py index 1ec2d9eab2076..368a0b1aee428 100644 --- a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py +++ b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py @@ -36,7 +36,7 @@ class BaseDagBundle(_BaseDagBundle): # type: ignore[no-redef] def _log(self): """Lazy structlog logger bound with common bundle context.""" if not hasattr(self, "_compat_structlog"): - log_context = self._log_context() if hasattr(self, "_log_context") else {} + log_context = self._get_log_context() if hasattr(self, "_get_log_context") else {} self._compat_structlog = structlog.get_logger(type(self).__module__).bind( bundle_name=self.name, version=self.version, diff --git a/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py index 0b4a84e3b0523..32c796107d5cc 100644 --- a/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py +++ b/providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py @@ -16,15 +16,52 @@ # under the License. from __future__ import annotations +import importlib +from pathlib import Path from unittest import mock import pytest +import structlog.testing pytest.importorskip("airflow.dag_processing.bundles", reason="Requires Airflow 3+") from airflow.dag_processing.bundles.base import BaseDagBundle as CoreBaseDagBundle from airflow.providers.common.compat.bundles import BaseDagBundle +from tests_common.test_utils.config import conf_vars + + +def _make_bundle_class(base): + """Return a concrete bundle subclass with the given base.""" + + class _DummyBundle(base): + @property + def path(self) -> Path: + return Path("/tmp") + + def initialize(self) -> None: + pass + + def get_current_version(self): + return None + + def refresh(self) -> None: + pass + + return _DummyBundle + + +@pytest.fixture +def legacy_compat_base(): + """Yield the compat BaseDagBundle as it would appear on an older Airflow without _log.""" + import airflow.providers.common.compat.bundles as mod + + with mock.patch.object(CoreBaseDagBundle, "_log", None, create=True): + importlib.reload(mod) + yield mod.BaseDagBundle + # Reload again outside the patch so the real module is back in sys.modules. + importlib.reload(mod) + class TestDagBundleCompat: def test_base_dag_bundle_is_importable(self): @@ -38,92 +75,29 @@ def test_has_log_property(self): def test_log_property_on_subclass_instance(self, tmp_path): """_log returns a bound structlog logger on concrete bundle subclasses.""" - from pathlib import Path - - from tests_common.test_utils.config import conf_vars - - class _DummyBundle(BaseDagBundle): - @property - def path(self) -> Path: - return tmp_path - - def initialize(self) -> None: - pass - - def get_current_version(self): - return None - - def refresh(self) -> None: - pass - + DummyBundle = _make_bundle_class(BaseDagBundle) with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): - bundle = _DummyBundle(name="test-bundle") + bundle = DummyBundle(name="test-bundle") - log = bundle._log - assert log is not None - # Calling _log twice returns the same (cached) logger. - assert bundle._log is log + first = bundle._log + assert first is not None + assert bundle._log is first # cached def test_log_is_bound_with_context(self, tmp_path): """_log includes bundle_name and version in its bound variables.""" - from pathlib import Path - - from tests_common.test_utils.config import conf_vars - - class _DummyBundle(BaseDagBundle): - @property - def path(self) -> Path: - return tmp_path - - def initialize(self) -> None: - pass - - def get_current_version(self): - return None - - def refresh(self) -> None: - pass - + DummyBundle = _make_bundle_class(BaseDagBundle) with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): - bundle = _DummyBundle(name="my-bundle", version="abc123") + bundle = DummyBundle(name="my-bundle", version="abc123") - log = bundle._log - # structlog bound loggers expose their bindings via _context - ctx = log._context if hasattr(log, "_context") else {} - assert ctx.get("bundle_name") == "my-bundle" - assert ctx.get("version") == "abc123" + with structlog.testing.capture_logs() as cap: + bundle._log.info("test event") + assert cap[0]["bundle_name"] == "my-bundle" + assert cap[0]["version"] == "abc123" - def test_compat_subclass_provides_log_when_missing(self, tmp_path): + def test_compat_subclass_provides_log_when_missing(self, tmp_path, legacy_compat_base): """Even when the installed BaseDagBundle has no _log, the compat class fills it in.""" - from pathlib import Path - - from tests_common.test_utils.config import conf_vars - - # Simulate an older Airflow version by temporarily hiding _log from - # the core class; the compat class must still expose it. - with mock.patch.object(CoreBaseDagBundle, "_log", None, create=True): - from importlib import reload - - import airflow.providers.common.compat.bundles as mod - - reload(mod) - CompatBase = mod.BaseDagBundle - - class _DummyBundle(CompatBase): - @property - def path(self) -> Path: - return tmp_path - - def initialize(self) -> None: - pass - - def get_current_version(self): - return None - - def refresh(self) -> None: - pass - + DummyBundle = _make_bundle_class(legacy_compat_base) with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): - bundle = _DummyBundle(name="legacy-bundle") + bundle = DummyBundle(name="legacy-bundle") assert bundle._log is not None From 5d34bef03a6fea481e071481aada836ac62e594e Mon Sep 17 00:00:00 2001 From: David Blain Date: Thu, 16 Jul 2026 21:47:28 +0200 Subject: [PATCH 7/7] Added missing setter on log for backward compatible BaseDagBundle in common.compat --- .../src/airflow/providers/common/compat/bundles/__init__.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py index 368a0b1aee428..73c98d4d3e24b 100644 --- a/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py +++ b/providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py @@ -44,5 +44,9 @@ def _log(self): ) return self._compat_structlog + @_log.setter + def _log(self, value) -> None: + self._compat_structlog = value + __all__ = ["BaseDagBundle"]