Dagster deployment fixes

opensource-observer · Dec 28, 2024 · 7abe9a5 · 7abe9a5
1 parent 0583026
commit 7abe9a5
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 10 deletions.
diff --git a/warehouse/oso_dagster/definitions.py b/warehouse/oso_dagster/definitions.py
@@ -65,13 +65,24 @@ def load_definitions():
         path=constants.sqlmesh_dir, gateway=constants.sqlmesh_gateway
     )
 
+    sqlmesh_infra_config = {
+        "environment": "prod",
+        "mcs_deployment_name": "production-mcs",
+        "mcs_deployment_namespace": "production-mcs",
+        "trino_deployment_namespace": "production-trino",
+        "trino_service_name": "production-trino-trino",
+        "trino_coordinator_deployment_name": "production-trino-trino-coordinator",
+        "trino_worker_deployment_name": "production-trino-trino-worker",
+    }
+
     early_resources = dict(
         project_id=project_id,
         staging_bucket=constants.staging_bucket,
         dlt_staging_destination=dlt_staging_destination,
         dlt_warehouse_destination=dlt_warehouse_destination,
         secrets=secret_resolver,
         sqlmesh_config=sqlmesh_config,
+        sqlmesh_infra_config=sqlmesh_infra_config,
     )
 
     asset_factories = load_all_assets_from_package(assets, early_resources)
@@ -110,15 +121,7 @@ def load_definitions():
         "project_id": project_id,
         "alert_manager": alert_manager,
         "sqlmesh_config": sqlmesh_config,
-        "sqlmesh_infra_config": {
-            "environment": "prod",
-            "mcs_deployment_name": "production-mcs",
-            "mcs_deployment_namespace": "production-mcs",
-            "trino_deployment_namespace": "production-trino",
-            "trino_service_name": "production-trino-trino",
-            "trino_coordinator_deployment_name": "production-trino-trino-coordinator",
-            "trino_worker_deployment_name": "production-trino-trino-worker",
-        },
+        "sqlmesh_infra_config": sqlmesh_infra_config,
         "sqlmesh": SQLMeshResource(config=sqlmesh_config),
     }
     for target in constants.main_dbt_manifests:

diff --git a/warehouse/oso_dagster/utils/asynctools.py b/warehouse/oso_dagster/utils/asynctools.py
@@ -5,6 +5,9 @@
 def safe_async_run[T](coro: t.Coroutine[t.Any, t.Any, T]) -> T:
     try:
         loop = asyncio.get_running_loop()
-        return loop.run_until_complete(coro)
     except RuntimeError:
+        loop = None
+
+    if loop is None:
         return asyncio.run(coro)
+    return loop.run_until_complete(coro)