Source code for pudl.output.sql.helpers
"""Helper functions for creating output assets."""
import importlib.resources
from dagster import AssetsDefinition, asset
[docs]
def sql_asset_factory(
name: str,
deps: set[str] = {},
io_manager_key: str = "pudl_io_manager",
compute_kind: str = "SQL",
) -> AssetsDefinition:
"""Factory for creating assets that run SQL statements."""
@asset(
name=name,
deps=deps,
io_manager_key=io_manager_key,
compute_kind=compute_kind,
)
def sql_view_asset() -> str:
"""Asset that creates sql view in a database."""
sql_path_traversable = (
importlib.resources.files("pudl.output.sql") / f"{name}.sql"
)
try:
with importlib.resources.as_file(sql_path_traversable) as sql_path:
return sql_path.read_text()
# Raise a helpful error here if a sql file doesn't exist
except FileNotFoundError as err:
raise FileNotFoundError(
f"Could not find {sql_path}. "
f"Create a sql file in pudl.output.sql subpackage for {name} asset."
) from err
return sql_view_asset