Asset factory - build a set of assets from YAML specs

[sryza]

from dagster import Definitions, AssetsDefinition, asset

specs = [
    {"name": "asset1", "upstream": [], "sql": "blah blah"},
    {"name": "asset2", "upstream": ["asset1"], "sql": "blah blah"},
]


def execute_sql(sql: str) -> None:
    ...


def build_asset(spec) -> AssetsDefinition:
    @asset(name=spec["name"], non_argument_deps=set(spec["upstreams"]))
    def _asset():
        execute_sql(spec["sql"])

    return _asset


defs = Definitions(assets=[build_asset(spec) for spec in specs])

[jburnich]

Hi!
The idea is interesting. However, I slightly modified the input configuration with this :

specs = [
    {"name": "asset1", "upstream": [], "sql": "query1"},
    {"name": "asset2", "upstream": ["asset1"], "sql": "query2"},
]

In both assets I found that query_2 was executed. I find it hard to understand why this behavior.

[sryza]

ooo this is a peculiarity with how Python handles closures inside of for loops. I'll post an update that fixes it.

[sryza]

Updated

[slopp]

Another example that shows how you might mix an asset factory with regular asset definitions:

from dagster import asset, Definitions, AssetIn

assets_to_make = ["a", "b", "c"]

def make_assets(asset_to_make):
    @asset(
        name=asset_to_make
    )
    def asset_template():
        print(asset_to_make)
        return 

    return asset_template

factory_assets = [make_assets(key) for key in assets_to_make]


factory_assets_ins = dict(zip(assets_to_make, [AssetIn(i) for i in assets_to_make]))

@asset
def downstream_of_factory_one(a):
    pass

@asset(
    ins=factory_assets_ins
)
def downstream_of_factory_all(**kwargs):
    pass

defs = Definitions(
    assets=[*factory_assets, downstream_of_factory_one, downstream_of_factory_all]
)

[bendnorman]

We're doing something similar but reading sql from files and executing it using an IO Manager:

def sql_asset_factory(
    name: str,
    non_argument_deps: set[str] = {},
    io_manager_key: str = "pudl_sqlite_io_manager",
    compute_kind: str = "SQL",
) -> AssetsDefinition:
    """Factory for creating assets that run SQL statements."""

    @asset(
        name=name,
        non_argument_deps=non_argument_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").joinpath(
            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:
            raise FileNotFoundError(
                f"Could not find {sql_path}. Create a sql file in pudl.output.sql subpackage for {name} asset."
            )

    return sql_view_asset

https://github.com/catalyst-cooperative/pudl/blob/927ec730dc24639b3a842eef3c97f3fffe6d3851/src/pudl/output/sql/helpers.py#L7-L35

[nitindatta]

hi @bendnorman that's interesting way, just one query if you add new asset in db, how does that get loaded, does it requires re-starting dagster

[zyd14]

@nitindatta This would require restarting Dagster to force re-querying the database for the SQL queries to generate assets for. Assets must be defined when code is loaded, at the global level, so asset factories require restarting Dagster to force the asset factory function to execute again

[garethbrickman]

We also have this YAML DSL for Asset Graphs Example

[Erin-Boehmer]

How would I trigger these assets to conditionally materialize using RunRequests from a single sensor?

✏️ Use case:

I have a sensor that detects whether a 3rd party JSON file has been updated. This JSON file is essentially a changelog for reports that we ingest. If a report has changes listed in the JSON file, we want to trigger an updated asset materialization for that report. I want to create the report assets via a factory pattern like yall show here, given we have hundreds. How do I get the sensor to trigger only those assets that have changes?

💭 Thoughts thus far:

It looks like both Sensors and RunRequests share the asset_selection parameter (docs). I'm not sure if my sensor should be triggering a job or asset materializations directly. And, if it is triggering a job, can I have a single job where the sensor passes in which assets are selected for materialization?

[garethbrickman]

Check out this discussion to see examples of how a sensor can materialize assets without specifying a job to do it: #23515