Use structlog for GitDagBundle (apache#47685)

When working on stale bundle cleanup I experimented with structlog and added some logging that used its bind feature.  I later removed the structlog part of it because I was not sure it would make it into 3.0. To keep the debug key-values I added a helper.  We can remove it now that structlog is expected to be in 3.0.

Author: dstandish

airflow/dag_processing/bundles/git.py

@@ -19,13 +19,13 @@
 
 import contextlib
 import json
-import logging
 import os
 import tempfile
 from pathlib import Path
 from typing import Any
 from urllib.parse import urlparse
 
+import structlog
 from git import Repo
 from git.exc import BadName, GitCommandError, NoSuchPathError
 
@@ -34,9 +34,8 @@
 )
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
-from airflow.utils.log.logging_mixin import LoggingMixin
 
-log = logging.getLogger(__name__)
+log = structlog.get_logger()
 
 
 class GitHook(BaseHook):
@@ -119,7 +118,7 @@ def configure_hook_env(self):
             yield
 
 
-class GitDagBundle(BaseDagBundle, LoggingMixin):
+class GitDagBundle(BaseDagBundle):
     """
     git DAG bundle - exposes a git repository as a DAG bundle.
 
@@ -154,33 +153,23 @@ def __init__(
         self.git_conn_id = git_conn_id
         self.repo_url = repo_url
 
-        def log_debug(msg, **kwargs):
-            if not log.isEnabledFor(logging.DEBUG):
-                return
-            # ugly; replace when structlog implemented
-            context = dict(
-                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,
-                repo_url=self.repo_url,
-            )
-            context.update(kwargs)
-
-            for k, v in context.items():
-                msg += f" {k}='{v}'"
-            log.debug(msg)
-
-        self._log_debug = log_debug
-        log_debug("bundle configured")
+        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,
+            repo_url=self.repo_url,
+        )
+
+        self._log.debug("bundle configured")
         try:
             self.hook = GitHook(git_conn_id=self.git_conn_id, repo_url=self.repo_url)
             self.repo_url = self.hook.repo_url
-            log_debug("repo_url updated from hook", repo_url=self.repo_url)
+            self._log.debug("repo_url updated from hook", repo_url=self.repo_url)
         except AirflowException as e:
-            self.log.warning("Could not create GitHook for connection %s : %s", self.git_conn_id, e)
+            self._log.warning("Could not create GitHook", conn_id=self.git_conn_id, exc=e)
 
     def _initialize(self):
         with self.lock():
@@ -190,7 +179,7 @@ def _initialize(self):
 
             self._clone_repo_if_required()
             self.repo.git.checkout(self.tracking_ref)
-            self._log_debug("bundle initialize", version=self.version)
+            self._log.debug("bundle initialize", version=self.version)
             if self.version:
                 if not self._has_version(self.repo, self.version):
                     self.repo.remotes.origin.fetch()
@@ -207,7 +196,7 @@ def initialize(self) -> None:
 
     def _clone_repo_if_required(self) -> None:
         if not os.path.exists(self.repo_path):
-            self.log.info("Cloning repository to %s from %s", self.repo_path, self.bare_repo_path)
+            self._log.info("Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path)
             try:
                 Repo.clone_from(
                     url=self.bare_repo_path,
@@ -217,14 +206,14 @@ def _clone_repo_if_required(self) -> None:
                 # Protection should the bare repo be removed manually
                 raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
         else:
-            self._log_debug("repo exists", repo_path=self.repo_path)
+            self._log.debug("repo exists", repo_path=self.repo_path)
         self.repo = Repo(self.repo_path)
 
     def _clone_bare_repo_if_required(self) -> None:
         if not self.repo_url:
             raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
         if not os.path.exists(self.bare_repo_path):
-            self.log.info("Cloning bare repository to %s", self.bare_repo_path)
+            self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
             try:
                 Repo.clone_from(
                     url=self.repo_url,

tests/dag_processing/bundles/test_git.py

@@ -20,6 +20,7 @@
 import os
 import re
 from unittest import mock
+from unittest.mock import patch
 
 import pytest
 from git import Repo
@@ -564,16 +565,20 @@ def test_clone_repo_no_such_path_error(self, mock_githook):
 
                 assert "Repository path: %s not found" in str(exc_info.value)
 
-    @mock.patch("airflow.dag_processing.bundles.git.GitDagBundle.log")
-    def test_repo_url_access_missing_connection_doesnt_error(self, mock_log):
+    @patch.dict(os.environ, {"AIRFLOW_CONN_MY_TEST_GIT": '{"host": "something"}'})
+    @pytest.mark.parametrize("conn_id, should_find", [("my_test_git", True), ("something-else", False)])
+    def test_repo_url_access_missing_connection_doesnt_error(self, conn_id, should_find):
         bundle = GitDagBundle(
             name="testa",
             tracking_ref="main",
-            git_conn_id="unknown",
+            git_conn_id=conn_id,
             repo_url="some_repo_url",
         )
         assert bundle.repo_url == "some_repo_url"
-        assert "Could not create GitHook for connection" in mock_log.warning.call_args[0][0]
+        if should_find:
+            assert isinstance(bundle.hook, GitHook)
+        else:
+            assert not hasattr(bundle, "hook")
 
     @mock.patch("airflow.dag_processing.bundles.git.GitHook")
     def test_lock_used(self, mock_githook, git_repo):