How to create dynamic specs for a dagster asset factory?

[noahtf13]

We are querying to get a list of tables we would like to monitor and then would like to build an asset for each item in this list. We run into a lot of issues since this involves top-level code.

The solutions such as sryza's asset factory (below) are great for predefined "specs" but they run into issues when we are trying to be more dynamic.

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])

Right now our pattern is

table_list = get_list_of_tables() # similar to specs in the example above

# want to create a downstream asset for each string (table name) in the list w/o top level code

Is there a way to avoid top-level code here? Can an asset that outputs a list be used to defined dynamic downstream assets?

[chrishiste]

Would also love this