From ab560e341d88f4631ef93ca6a40b64f080fb0d07 Mon Sep 17 00:00:00 2001
From: Dazhong Xia <dazhong.xia@catalyst.coop>
Date: Fri, 13 Dec 2024 14:21:21 -0500
Subject: [PATCH] WIP: Use same dagster Definitions for test and prod.

---
 src/pudl/etl/__init__.py |  2 --
 src/pudl/etl/cli.py      | 20 +++-----------------
 2 files changed, 3 insertions(+), 19 deletions(-)

diff --git a/src/pudl/etl/__init__.py b/src/pudl/etl/__init__.py
index 475c38970d..016110a959 100644
--- a/src/pudl/etl/__init__.py
+++ b/src/pudl/etl/__init__.py
@@ -34,8 +34,6 @@
 from pudl.settings import EtlSettings
 
 from . import (
-    check_foreign_keys,
-    cli,
     eia_bulk_elec_assets,
     epacems_assets,
     glue_assets,
diff --git a/src/pudl/etl/cli.py b/src/pudl/etl/cli.py
index 016c12ff50..c4018bf446 100644
--- a/src/pudl/etl/cli.py
+++ b/src/pudl/etl/cli.py
@@ -8,14 +8,13 @@
 import fsspec
 from dagster import (
     DagsterInstance,
-    Definitions,
     JobDefinition,
     build_reconstructable_job,
-    define_asset_job,
     execute_job,
 )
 
 import pudl
+from pudl.etl import defs
 from pudl.helpers import get_dagster_execution_config
 from pudl.settings import EpaCemsSettings, EtlSettings
 from pudl.workspace.setup import PudlPaths
@@ -40,21 +39,8 @@ def pudl_etl_job_factory(
     def get_pudl_etl_job():
         """Create an pudl_etl_job wrapped by to be wrapped by reconstructable."""
         pudl.logging_helpers.configure_root_logger(logfile=logfile, loglevel=loglevel)
-        jobs = [define_asset_job("etl_job")]
-        if not process_epacems:
-            jobs = [
-                define_asset_job(
-                    "etl_job",
-                    selection=pudl.etl.create_non_cems_selection(
-                        pudl.etl.default_assets
-                    ),
-                )
-            ]
-        return Definitions(
-            assets=pudl.etl.default_assets,
-            resources=pudl.etl.default_resources,
-            jobs=jobs,
-        ).get_job_def("etl_job")
+        job_name = "etl_full_no_cems" if not process_epacems else "etl_full"
+        return defs.get_job_def(job_name)
 
     return get_pudl_etl_job