Allow specifying repo_url in GitDagBundle kwargs (apache#46345)

Author: jx2lee

airflow/dag_processing/bundles/git.py

@@ -98,6 +98,7 @@ class GitDagBundle(BaseDagBundle, LoggingMixin):
     :param tracking_ref: Branch or tag for this DAG bundle
     :param subdir: Subdirectory within the repository where the DAGs are stored (Optional)
     :param git_conn_id: Connection ID for SSH/token based connection to the repository (Optional)
+    :param repo_url: Explicit Git repository URL to override the connection's host. (Optional)
     """
 
     supports_versioning = True
@@ -108,6 +109,7 @@ def __init__(
         tracking_ref: str,
         subdir: str | None = None,
         git_conn_id: str = "git_default",
+        repo_url: str | None = None,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -118,10 +120,9 @@ def __init__(
             self._dag_bundle_root_storage_path / "git" / (self.name + f"+{self.version or self.tracking_ref}")
         )
         self.git_conn_id = git_conn_id
-        self.repo_url: str | None = None
         try:
             self.hook = GitHook(git_conn_id=self.git_conn_id)
-            self.repo_url = self.hook.repo_url
+            self.repo_url = repo_url or self.hook.repo_url
         except AirflowException as e:
             self.log.error("Error creating GitHook: %s", e)
 

tests/dag_processing/test_dag_bundles.py

@@ -207,6 +207,21 @@ def test_uses_dag_bundle_root_storage_path(self, git_repo):
         bundle = GitDagBundle(name="test", tracking_ref=GIT_DEFAULT_BRANCH)
         assert str(bundle._dag_bundle_root_storage_path) in str(bundle.path)
 
+    def test_repo_url_overrides_connection_host_when_provided(self, git_repo):
+        repo_path, repo = git_repo
+        bundle = GitDagBundle(
+            name="test", tracking_ref=GIT_DEFAULT_BRANCH, repo_url="git@myorg.github.com:apache/airflow.git"
+        )
+        assert bundle.repo_url != bundle.hook.repo_url
+        assert bundle.repo_url == "git@myorg.github.com:apache/airflow.git"
+
+    def test_falls_back_to_connection_host_when_no_repo_url_provided(self, git_repo):
+        repo_path, repo = git_repo
+        bundle = GitDagBundle(name="test", tracking_ref=GIT_DEFAULT_BRANCH)
+
+        assert bundle.repo_url
+        assert bundle.repo_url == bundle.hook.repo_url
+
     @mock.patch("airflow.dag_processing.bundles.git.GitHook")
     def test_get_current_version(self, mock_githook, git_repo):
         repo_path, repo = git_repo