Bring the Census DP1 to SQLite ETL into dagster

src/pudl/census_to_sqlite/__init__.py

@@ -0,0 +1,29 @@
+"""Dagster definitions for the FERC to SQLite process."""
+
+from dagster import Definitions, graph
+
+import pudl
+from pudl.convert.censusdp1tract_to_sqlite import censusdp1tract_to_sqlite
+from pudl.resources import datastore
+
+logger = pudl.logging_helpers.get_logger(__name__)
+
+
+@graph
+def census_to_sqlite():
+    """Clone the Census DP1 database into SQLite."""
+    censusdp1tract_to_sqlite()
+
+
+default_resources_defs = {
+    "datastore": datastore,
+}
+
+census_to_sqlite = census_to_sqlite.to_job(
+    resource_defs=default_resources_defs,
+    name="census_to_sqlite",
+)
+
+defs: Definitions = Definitions(jobs=[census_to_sqlite])
+"""A collection of dagster assets, resources, IO managers, and jobs for the FERC to
+SQLite ETL."""

src/pudl/census_to_sqlite/cli.py

@@ -0,0 +1,147 @@
+"""A script for cloning the Census DP1 database into SQLite.
+
+This script generates a SQLite database that is a clone/mirror of the original
+Census DP1 database. We use this cloned database as the starting point for the
+main PUDL ETL process. The underlying work in the script is being done in
+:mod:`pudl.extract.ferc1`.
+"""
+import argparse
+import sys
+from collections.abc import Callable
+
+from dagster import (
+    DagsterInstance,
+    JobDefinition,
+    build_reconstructable_job,
+    execute_job,
+)
+
+import pudl
+from pudl import census_to_sqlite
+from pudl.settings import EtlSettings
+
+# Create a logger to output any messages we might have...
+logger = pudl.logging_helpers.get_logger(__name__)
+
+
+def parse_command_line(argv):
+    """Parse command line arguments. See the -h option.
+
+    Args:
+        argv (str): Command line arguments, including caller filename.
+
+    Returns:
+        dict: Dictionary of command line arguments and their parsed values.
+    """
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "settings_file", type=str, default="", help="path to YAML settings file."
+    )
+    parser.add_argument(
+        "--logfile",
+        default=None,
+        type=str,
+        help="If specified, write logs to this file.",
+    )
+    parser.add_argument(
+        "-c",
+        "--clobber",
+        action="store_true",
+        help="""Clobber existing sqlite database if it exists. If clobber is
+        not included but the sqlite databse already exists the _build will
+        fail.""",
+        default=False,
+    )
+    parser.add_argument(
+        "--sandbox",
+        action="store_true",
+        default=False,
+        help="Use the Zenodo sandbox rather than production",
+    )
+    parser.add_argument(
+        "--gcs-cache-path",
+        type=str,
+        help="Load datastore resources from Google Cloud Storage. Should be gs://bucket[/path_prefix]",
+    )
+    parser.add_argument(
+        "--loglevel",
+        help="Set logging level (DEBUG, INFO, WARNING, ERROR, or CRITICAL).",
+        default="INFO",
+    )
+    arguments = parser.parse_args(argv[1:])
+    return arguments
+
+
+def census_to_sqlite_job_factory(
+    logfile: str | None = None, loglevel: str = "INFO"
+) -> Callable[[], JobDefinition]:
+    """Factory for parameterizing a reconstructable census_to_sqlite job.
+
+    Args:
+        loglevel: The log level for the job's execution.
+        logfile: Path to a log file for the job's execution.
+
+    Returns:
+        The job definition to be executed.
+    """
+
+    def get_census_to_sqlite_job():
+        """Module level func for creating a job to be wrapped by reconstructable."""
+        return census_to_sqlite.census_to_sqlite.to_job(
+            resource_defs=census_to_sqlite.default_resources_defs,
+            name="census_to_sqlite_job",
+        )
+
+    return get_census_to_sqlite_job
+
+
+def main():  # noqa: C901
+    """Clone the Census database into SQLite."""
+    args = parse_command_line(sys.argv)
+
+    # Display logged output from the PUDL package:
+    pudl.logging_helpers.configure_root_logger(
+        logfile=args.logfile, loglevel=args.loglevel
+    )
+
+    etl_settings = EtlSettings.from_yaml(args.settings_file)
+
+    # Set PUDL_INPUT/PUDL_OUTPUT env vars from .pudl.yml if not set already!
+    pudl.workspace.setup.get_defaults()
+
+    census_to_sqlite_reconstructable_job = build_reconstructable_job(
+        "pudl.census_to_sqlite.cli",
+        "census_to_sqlite_job_factory",
+        reconstructable_kwargs={"loglevel": args.loglevel, "logfile": args.logfile},
+    )
+
+    result = execute_job(
+        census_to_sqlite_reconstructable_job,
+        instance=DagsterInstance.get(),
+        run_config={
+            "resources": {
+                "census_to_sqlite_settings": {
+                    "config": etl_settings.census_to_sqlite_settings.dict()
+                },
+                "datastore": {
+                    "config": {
+                        "sandbox": args.sandbox,
+                        "gcs_cache_path": args.gcs_cache_path
+                        if args.gcs_cache_path
+                        else "",
+                    },
+                },
+            },
+        },
+        raise_on_error=True,
+    )
+
+    # Workaround to reliably getting full stack trace
+    if not result.success:
+        for event in result.all_events:
+            if event.event_type_value == "STEP_FAILURE":
+                raise Exception(event.event_specific_data.error)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/pudl/convert/censusdp1tract_to_sqlite.py

@@ -21,13 +21,36 @@
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
+from dagster import Field, op
+
 import pudl
+from pudl.helpers import EnvVar
 from pudl.workspace.datastore import Datastore
 
 logger = pudl.logging_helpers.get_logger(__name__)
 
 
-def censusdp1tract_to_sqlite(pudl_settings=None, year=2010, ds=None, clobber=False):
+@op(
+    config_schema={
+        "pudl_output_path": Field(
+            EnvVar(
+                env_var="PUDL_OUTPUT",
+            ),
+            description="Path of directory to store the database in.",
+            default_value=None,
+        ),
+        "clobber": Field(
+            bool, description="Clobber existing Census database.", default_value=False
+        ),
+        "year": Field(
+            int,
+            description="Year of Census data to extract (currently must be 2010).",
+            default_value=2010,
+        ),
+    },
+    required_resource_keys={"datastore"},
+)
+def censusdp1tract_to_sqlite(context):
     """Use GDAL's ogr2ogr utility to convert the Census DP1 GeoDB to an SQLite DB.
 
     The Census DP1 GeoDB is read from the datastore, where it is stored as a
@@ -36,13 +59,10 @@ def censusdp1tract_to_sqlite(pudl_settings=None, year=2010, ds=None, clobber=Fal
     resulting SQLite DB file is put in the PUDL output directory alongside the
     ferc1 and pudl SQLite databases.
 
-    Args:
-        pudl_settings (dict): A PUDL settings dictionary.
-        year (int): Year of Census data to extract (currently must be 2010)
-
     Returns:
         None
     """
+    ds = context.resources.datastore
     if ds is None:
         ds = Datastore()
     # If we're in a conda environment, use the version of ogr2ogr that has been
@@ -60,12 +80,14 @@ def censusdp1tract_to_sqlite(pudl_settings=None, year=2010, ds=None, clobber=Fal
     with TemporaryDirectory() as tmpdir:
         # Use datastore to grab the Census DP1 zipfile
         tmpdir_path = Path(tmpdir)
-        zip_ref = ds.get_zipfile_resource("censusdp1tract", year=year)
+        zip_ref = ds.get_zipfile_resource(
+            "censusdp1tract", year=context.op_config["year"]
+        )
         extract_root = tmpdir_path / Path(zip_ref.filelist[0].filename)
-        out_path = Path(pudl_settings["pudl_out"]) / "censusdp1tract.sqlite"
+        out_path = Path(context.op_config["pudl_output_path"]) / "censusdp1tract.sqlite"
 
         if out_path.exists():
-            if clobber:
+            if context.op_config["clobber"]:
                 out_path.unlink()
             else:
                 raise SystemExit(

src/pudl/etl/__init__.py

@@ -39,6 +39,9 @@
     *load_assets_from_modules([pudl.transform.ferc1], group_name="norm_ferc1"),
     *load_assets_from_modules([pudl.extract.ferc714], group_name="raw_ferc714"),
     *load_assets_from_modules([pudl.transform.ferc714], group_name="clean_ferc714"),
+    *load_assets_from_modules(
+        [pudl.convert.censusdp1tract_to_sqlite], group_name="censusdp1"
+    ),
     *load_assets_from_modules([glue_assets], group_name="glue"),
     *load_assets_from_modules([static_assets], group_name="static"),
     *load_assets_from_modules(