Merge pull request #217 from DevDataPlatform/216-pass-the-bigquery-lo…

…cation-to-prefect-proxy-for-dbt-core-blocks

pass bigquery dataset location to prefec-proxy for dbt block creation

ddpui/api/client/prefect_api.py

@@ -355,6 +355,15 @@ def post_prefect_dbt_core_block(request, payload: PrefectDbtRun):
         else orguser.org.dbt.default_schema
     )
 
+    # get the bigquery location if warehouse is bq
+    bqlocation = None
+    if warehouse.wtype == "bigquery":
+        destination = airbyte_service.get_destination(
+            orguser.org.airbyte_workspace_id, warehouse.airbyte_destination_id
+        )
+        if destination.get("connectionConfiguration"):
+            bqlocation = destination["connectionConfiguration"]["dataset_location"]
+
     block_names = []
     for sequence_number, command in enumerate(
         ["clean", "deps", "run", "test", "docs generate"]
@@ -381,6 +390,7 @@ def post_prefect_dbt_core_block(request, payload: PrefectDbtRun):
                 target,
                 warehouse.wtype,
                 credentials,
+                bqlocation,
             )
         except Exception as error:
             logger.exception(error)

ddpui/ddpprefect/prefect_service.py

@@ -201,6 +201,7 @@ def create_dbt_core_block(
     target: str,
     wtype: str,
     credentials: dict,
+    bqlocation: str,
 ) -> dict:
     """Create a dbt core block in prefect"""
     response = prefect_post(
@@ -214,6 +215,7 @@ def create_dbt_core_block(
             },
             "wtype": wtype,
             "credentials": credentials,
+            "bqlocation": bqlocation,
             "commands": dbtcore.commands,
             "env": dbtcore.env,
             "working_dir": dbtcore.working_dir,