Simplify materialize_asset.py script

devtools/materialize_asset.py

@@ -1,13 +1,21 @@
 #! /usr/bin/env python
-"""Materialize one asset & its upstream deps in-process so you can debug."""
+"""Materialize one asset & its upstream deps in-process so you can debug.
+
+If you are using the VSCode Debugger, you'll need to specify the asset_id
+in the launch.json file:
+
+{
+    ...,
+    "args": ["{YOUR_ASSET_ID}}"]
+    ...,
+}
+"""
 
 import argparse
-import importlib.resources
 
-from dagster import AssetSelection, Definitions, define_asset_job
+from dagster import materialize
 
-from pudl import etl
-from pudl.settings import EtlSettings
+from pudl.etl import default_assets, default_resources, load_dataset_settings_from_file
 
 
 def _parse():
@@ -19,35 +27,23 @@ def _parse():
 def main(asset_id):
     """Entry point.
 
-    Defines dagster context like in etl/__init__.py - needs to be kept in sync.
+    Materialize one asset & its upstream deps in-process so you can debug.
 
-    Then creates a job with asset selection.
+    Args:
+        asset_id: Name of asset you want to materialize.
     """
-    pkg_source = importlib.resources.files("pudl.package_data.settings").joinpath(
-        "etl_fast.yml"
-    )
-    with importlib.resources.as_file(pkg_source) as yf:
-        etl_fast_settings = EtlSettings.from_yaml(yf).datasets
-
-    # TODO (daz/zach): maybe there's a way to do this directly with dagster cli?
-    defs = Definitions(
-        assets=etl.default_assets,
-        resources=etl.default_resources,
-        jobs=[
-            define_asset_job(
-                name="materialize_one",
-                selection=AssetSelection.keys(asset_id).upstream(),
-                config={
-                    "resources": {
-                        "dataset_settings": {
-                            "config": etl_fast_settings.dict(),
-                        },
-                    },
-                },
-            ),
-        ],
+    materialize(
+        default_assets,
+        selection=f"*{asset_id}",
+        resources=default_resources,
+        run_config={
+            "resources": {
+                "dataset_settings": {
+                    "config": load_dataset_settings_from_file("etl_fast")
+                }
+            }
+        },
     )
-    defs.get_job_def("materialize_one").execute_in_process()
 
 
 if __name__ == "__main__":