diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py b/airflow-core/src/airflow/dag_processing/bundles/base.py index 344a3349fecab..c4d0288474199 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: 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: 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/amazon/pyproject.toml b/providers/amazon/pyproject.toml index eddb2645e6080..cd3a7e9d3a91a 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 65bacb4b38811..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,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(): 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..73c98d4d3e24b --- /dev/null +++ b/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"] 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..32c796107d5cc --- /dev/null +++ b/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 diff --git a/providers/git/pyproject.toml b/providers/git/pyproject.toml index 7056a12e6b5cd..fcfa1d450def5 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 275aa6b3f634b..3f6033c813f4a 100644 --- a/providers/git/src/airflow/providers/git/bundles/git.py +++ b/providers/git/src/airflow/providers/git/bundles/git.py @@ -22,12 +22,11 @@ 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 @@ -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/pyproject.toml b/providers/google/pyproject.toml index 1698491c65ff8..77d350c311941 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 e9806276a823a..2a99ff0dede2e 100644 --- a/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py +++ b/providers/google/src/airflow/providers/google/cloud/bundles/gcs.py @@ -19,10 +19,9 @@ import os from pathlib import Path -import structlog 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 @@ -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():