Centralise structlog initialisation in BaseDagBundle

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: Any = None
         self.name = name
         self.version = version
         self.version_data = version_data
@@ -350,6 +352,24 @@ def initialize(self) -> None:
                 bundle_path,
             )
 
+    def _get_log_context(self) -> dict[str, Any]:
+        """Return extra structlog context fields for this bundle. Override in subclasses."""
+        return {}
+
+    @property
+    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._get_log_context()
+            )
+        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:

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

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

providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py

@@ -19,11 +19,9 @@
 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
+from airflow.providers.common.compat.bundles import BaseDagBundle
 from airflow.providers.common.compat.sdk import AirflowException
 
 
@@ -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():

providers/common/compat/src/airflow/providers/common/compat/bundles/__init__.py

@@ -0,0 +1,52 @@
+# 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 inspect
+
+import structlog
+
+from airflow.dag_processing.bundles.base import BaseDagBundle as _BaseDagBundle
+
+if isinstance(inspect.getattr_static(_BaseDagBundle, "_log", None), property):
+    # 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"):
+                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,
+                    **log_context,
+                )
+            return self._compat_structlog
+
+        @_log.setter
+        def _log(self, value) -> None:
+            self._compat_structlog = value
+
+
+__all__ = ["BaseDagBundle"]

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.

providers/common/compat/tests/unit/common/compat/bundles/test_dag_bundle.py

@@ -0,0 +1,103 @@
+# 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 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):
+        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."""
+        DummyBundle = _make_bundle_class(BaseDagBundle)
+        with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}):
+            bundle = DummyBundle(name="test-bundle")
+
+        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."""
+        DummyBundle = _make_bundle_class(BaseDagBundle)
+        with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}):
+            bundle = DummyBundle(name="my-bundle", 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, legacy_compat_base):
+        """Even when the installed BaseDagBundle has no _log, the compat class fills it in."""
+        DummyBundle = _make_bundle_class(legacy_compat_base)
+        with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}):
+            bundle = DummyBundle(name="legacy-bundle")
+
+        assert bundle._log is not None

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",
 ]
 

providers/git/src/airflow/providers/git/bundles/git.py

@@ -22,21 +22,18 @@
 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
 
-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
 
 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:

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'",