attempt to dgify

opensource_mds/__init__.py

@@ -1,36 +0,0 @@
-from dagster import (
-    Definitions,
-    load_assets_from_modules,
-    define_asset_job,
-)
-
-from . import assets, resources
-
-all_assets = load_assets_from_modules([assets])
-duckdb_resource = resources.duckdb_resource
-sling_resource = resources.sling_resource
-dbt_resource = resources.dbt_resource
-
-assets_job = define_asset_job(
-    name="etl_job",
-    selection=all_assets,
-    config={
-        "execution": {
-            "config": {
-                "multiprocess": {
-                    "max_concurrent": 4,
-                },
-            }
-        }
-    },
-)
-
-defs = Definitions(
-    assets=all_assets,
-    resources={
-        "duckdb": duckdb_resource,
-        "sling": sling_resource,
-        "dbt": dbt_resource,
-    },
-    jobs=[assets_job],
-)

opensource_mds/definitions.py

@@ -0,0 +1,14 @@
+import dagster as dg
+import opensource_mds.defs as definitions
+from opensource_mds.defs import resources
+
+defs = dg.Definitions.merge(
+    dg.components.load_defs(definitions),
+    dg.Definitions(
+        resources={
+            "duckdb": resources.duckdb_resource,
+            "sling": resources.sling_resource,
+            "dbt": resources.dbt_resource,
+        },
+    ),
+)

opensource_mds/assets.py → opensource_mds/defs/assets.py

@@ -1,21 +1,20 @@
 import time
 import zipfile
 from tempfile import NamedTemporaryFile
-from typing import Any, List, Mapping, Tuple
-import dagster as dg
-
-from dagster_sling import DagsterSlingTranslator
-import requests
+from typing import List, Tuple
 
+import dagster as dg
 from dagster_dbt import DbtCliResource, dbt_assets
+import requests
 from dagster_duckdb import DuckDBResource
 from dagster_embedded_elt.sling import (
-    sling_assets,
     SlingResource,
+    sling_assets,
 )
 
+from opensource_mds.defs.resources import CustomDagsterDbtTranslator, dbt_manifest_path
+
 from . import constants
-from .resources import CustomDagsterDbtTranslator, dbt_manifest_path
 
 retry_policy = dg.RetryPolicy(
     max_retries=3,
@@ -226,22 +225,18 @@ def sites(context: dg.AssetExecutionContext, duckdb: DuckDBResource):
 replication_config = {
     "source": "postgres",
     "target": "duckdb",
-    "defaults": {
-        "mode": "full-refresh",
-        "object": "{stream_schema}_{stream_table}"
-    },
-    "streams": {
-        "public.tickets": None,
-        "public.events": None
-    }
+    "defaults": {"mode": "full-refresh", "object": "{stream_schema}_{stream_table}"},
+    "streams": {"public.tickets": None, "public.events": None},
 }
 
+
 @sling_assets(replication_config=replication_config)
 def sling_sync_assets(context, sling: SlingResource):
     yield from sling.replicate(context=context)
     for row in sling.stream_raw_logs():
         context.log.info(row)
 
+
 @dg.asset(
     deps=[sling_sync_assets],
     compute_kind="duckdb",
@@ -250,7 +245,9 @@ def sling_sync_assets(context, sling: SlingResource):
 )
 def tickets(context: dg.AssetExecutionContext, duckdb: DuckDBResource):
     with duckdb.get_connection() as conn:
-        conn.execute("CREATE OR REPLACE TABLE tickets AS (SELECT * FROM main.public_tickets)")
+        conn.execute(
+            "CREATE OR REPLACE TABLE tickets AS (SELECT * FROM main.public_tickets)"
+        )
         nrows = conn.execute("SELECT COUNT(*) FROM tickets").fetchone()[0]  # type: ignore
 
         metadata = conn.execute(
@@ -279,7 +276,9 @@ def tickets(context: dg.AssetExecutionContext, duckdb: DuckDBResource):
 )
 def events(context: dg.AssetExecutionContext, duckdb: DuckDBResource):
     with duckdb.get_connection() as conn:
-        conn.execute("CREATE OR REPLACE TABLE events AS (SELECT * FROM main.public_events)")
+        conn.execute(
+            "CREATE OR REPLACE TABLE events AS (SELECT * FROM main.public_events)"
+        )
         nrows = conn.execute("SELECT COUNT(*) FROM events ").fetchone()[0]  # type: ignore
 
         metadata = conn.execute(

opensource_mds/defs/constants.py

@@ -0,0 +1,11 @@
+CHECKLIST_2020 = "https://clo-pfw-prod.s3.us-west-2.amazonaws.com/data/202306/PFW_all_2016_2020_June2023_Public.zip"
+CHECKLIST_2023 = "https://clo-pfw-prod.s3.us-west-2.amazonaws.com/data/202306/PFW_all_2021_2023_June2023_Public.zip"
+SITE_DESCRIPTION_DATA = "https://clo-pfw-prod.s3.us-west-2.amazonaws.com/data/202306/PFW_count_site_data_public_June2023.zip"
+SPECIES_TRANSLATION_DATA = "https://clo-pfw-prod.s3.us-west-2.amazonaws.com/data/202306/PFW_spp_translation_table_May2023.zip"
+
+CL_2020_FPATH = "../data/raw/checklist_data/PFW_all_2016_2020_June2023_Public.csv"
+CL_2023_FPATH = "../data/raw/checklist_data/PFW_all_2021_2023_June2023_Public.csv"
+SITE_DATA_FPATH = "../data/raw/checklist_data/PFW_count_site_data_public_June2023.csv"
+SPECIES_TRANSLATION_FPATH = (
+    "../data/raw/checklist_data/PFW_spp_translation_table_May2023.csv"
+)

opensource_mds/resources.py → opensource_mds/defs/resources.py

@@ -1,5 +1,5 @@
 from dagster import file_relative_path, get_dagster_logger
-from dagster import AssetKey, EnvVar
+from dagster import AssetKey
 from dagster_dbt import DagsterDbtTranslator, DbtCliResource
 from dagster_duckdb import DuckDBResource
 from dagster_embedded_elt.sling import (
@@ -34,9 +34,9 @@
 )
 
 dbt_resource = DbtCliResource(
-    project_dir=file_relative_path(__file__, "../dbt_project")
+    project_dir=file_relative_path(__file__, "../../dbt_project")
 )
-dbt_parse_invocation = dbt_resource.cli(["parse"], manifest={}).wait()
+dbt_parse_invocation = dbt_resource.cli(["parse", "-q"], manifest={}).wait()
 dbt_manifest_path = dbt_parse_invocation.target_path.joinpath("manifest.json")
 
 

pyproject.toml

@@ -23,3 +23,11 @@ build-backend = "setuptools.build_meta"
 
 [tool.dagster]
 module_name = "opensource_mds"
+
+# Add dg configuration
+[tool.dg]
+directory_type = "project"
+
+[tool.dg.project]
+root_module = "opensource_mds"
+code_location_target_module = "opensource_mds.definitions"