From 8592c4c88a54bc920e1f637522c23a0a29993332 Mon Sep 17 00:00:00 2001
From: okayhooni <ok9897@gmail.com>
Date: Mon, 27 Feb 2023 23:59:12 +0900
Subject: [PATCH 1/2] handling project location param on async dts trigger

---
 .../providers/google/cloud/hooks/bigquery_dts.py    | 13 ++++++++++++-
 .../providers/google/cloud/triggers/bigquery_dts.py |  1 +
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/airflow/providers/google/cloud/hooks/bigquery_dts.py b/airflow/providers/google/cloud/hooks/bigquery_dts.py
index 5c7ec5bf0612f..044e89e6b5f65 100644
--- a/airflow/providers/google/cloud/hooks/bigquery_dts.py
+++ b/airflow/providers/google/cloud/hooks/bigquery_dts.py
@@ -304,11 +304,16 @@ async def _get_project_id(self) -> str:
         sync_hook = await self.get_sync_hook()
         return sync_hook.project_id
 
+    async def _get_project_location(self) -> str:
+        sync_hook = await self.get_sync_hook()
+        return sync_hook.location
+
     async def get_transfer_run(
         self,
         config_id: str,
         run_id: str,
         project_id: str | None,
+        location: str | None = None,
         retry: Retry | _MethodDefault = DEFAULT,
         timeout: float | None = None,
         metadata: Sequence[tuple[str, str]] = (),
@@ -321,6 +326,7 @@ async def get_transfer_run(
         :param project_id: The BigQuery project id where the transfer configuration should be
             created. If set to None or missing, the default project_id from the Google Cloud connection
             is used.
+        :param location: BigQuery Transfer Service location for regional transfers.
         :param retry: A retry object used to retry requests. If `None` is
             specified, requests will not be retried.
         :param timeout: The amount of time, in seconds, to wait for the request to
@@ -330,8 +336,13 @@ async def get_transfer_run(
         :return: An ``google.cloud.bigquery_datatransfer_v1.types.TransferRun`` instance.
         """
         project_id = project_id or (await self._get_project_id())
+        location = location or (await self._get_project_location())
+        name = f"projects/{project_id}"
+        if location:
+            name += f"/locations/{location}"
+        name += f"/transferConfigs/{config_id}/runs/{run_id}"
+
         client = await self._get_conn()
-        name = f"projects/{project_id}/transferConfigs/{config_id}/runs/{run_id}"
         transfer_run = await client.get_transfer_run(
             name=name,
             retry=retry,
diff --git a/airflow/providers/google/cloud/triggers/bigquery_dts.py b/airflow/providers/google/cloud/triggers/bigquery_dts.py
index 2b4e57e9d9c98..a16585aaf4cd4 100644
--- a/airflow/providers/google/cloud/triggers/bigquery_dts.py
+++ b/airflow/providers/google/cloud/triggers/bigquery_dts.py
@@ -101,6 +101,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                     project_id=self.project_id,
                     config_id=self.config_id,
                     run_id=self.run_id,
+                    location=self.location,
                 )
                 state = transfer_run.state
                 self.log.info("Current state is %s", state)

From cdde11a63f776d14a404763d04a0d057f9eea315 Mon Sep 17 00:00:00 2001
From: okayhooni <ok9897@gmail.com>
Date: Tue, 28 Feb 2023 00:22:16 +0900
Subject: [PATCH 2/2] handling templated requested_run_time ~ seconds (int)

---
 airflow/providers/google/cloud/operators/bigquery_dts.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/airflow/providers/google/cloud/operators/bigquery_dts.py b/airflow/providers/google/cloud/operators/bigquery_dts.py
index 9712658a76be0..d230e6c157aff 100644
--- a/airflow/providers/google/cloud/operators/bigquery_dts.py
+++ b/airflow/providers/google/cloud/operators/bigquery_dts.py
@@ -306,6 +306,10 @@ def hook(self) -> BiqQueryDataTransferServiceHook:
 
     def execute(self, context: Context):
         self.log.info("Submitting manual transfer for %s", self.transfer_config_id)
+
+        if self.requested_run_time and isinstance(self.requested_run_time.get("seconds"), str):
+            self.requested_run_time["seconds"] = int(self.requested_run_time["seconds"])
+
         response = self.hook.start_manual_transfer_runs(
             transfer_config_id=self.transfer_config_id,
             requested_time_range=self.requested_time_range,
