Prefer BigQueryInsertJobOperator's project_id over hook's project_id for openlineage (#55948)

* Prefer operator's project_id over hook's project_id

* Remove unused import

* Formatting

* Remove unnecessary arguments

* Make mypy happy

Author: BasPH

providers/google/src/airflow/providers/google/cloud/openlineage/mixins.py

@@ -97,7 +97,9 @@ def get_openlineage_facets_on_complete(self, _):
         run_facets: dict[str, RunFacet] = {
             "externalQuery": ExternalQueryRunFacet(externalQueryId=self.job_id, source="bigquery")
         }
-        self._client = self.hook.get_client(project_id=self.hook.project_id, location=self.location)
+        self._client = self.hook.get_client(
+            project_id=self.project_id or self.hook.project_id, location=self.location
+        )
         try:
             job_properties = self._client.get_job(job_id=self.job_id)._properties
 

providers/google/tests/unit/google/cloud/openlineage/test_mixins.py

@@ -1013,3 +1013,37 @@ def test_generate_column_lineage_facet(self):
                 ),
             }
         )
+
+    def test_project_id_selection(self):
+        """
+        Check if project_id set via an argument to the operator takes prevalence over project_id set in a
+        connection.
+        """
+        from airflow.providers.google.cloud.operators.cloud_base import GoogleCloudBaseOperator
+
+        class TestOperator(GoogleCloudBaseOperator, _BigQueryInsertJobOperatorOpenLineageMixin):
+            def __init__(self, project_id: str | None = None, **_):
+                self.project_id = project_id
+                self.job_id = "foobar"
+                self.location = "foobar"
+                self.sql = "foobar"
+
+        # First test task where project_id is set explicitly
+        test = TestOperator(project_id="project_a")
+        test.hook = MagicMock()
+        test.hook.project_id = "project_b"
+        test._client = MagicMock()
+
+        test.get_openlineage_facets_on_complete(None)
+        _, kwargs = test.hook.get_client.call_args
+        assert kwargs["project_id"] == "project_a"
+
+        # Then test task where project_id is inherited from the hook
+        test = TestOperator()
+        test.hook = MagicMock()
+        test.hook.project_id = "project_b"
+        test._client = MagicMock()
+
+        test.get_openlineage_facets_on_complete(None)
+        _, kwargs = test.hook.get_client.call_args
+        assert kwargs["project_id"] == "project_b"