How do I create a job that selects assets based on some metadata?

[gnilrets]

Suppose I have a set of assets

from dagster import Definitions, asset, define_asset_job

@asset(metadata={'odd': True})
def asset1():
    return 'one'


@asset
def asset2(asset1):
    return 'two'

@asset(metadata={'odd': True})
def asset3():
    return 'three'

I want create an asset job that just selects assets that have metadata matching metadata.get('odd', False) is True. How can I make that work?

[jamiedemaria]

It's not the prettiest, but you could do something like this

@asset(metadata={'odd': True})
def asset1():
    return 'one'

@asset
def asset2(asset1):
    return 'two'

@asset(metadata={'odd': True})
def asset3():
    return 'three'

all_assets = [asset1, asset2, asset3]
odd_assets = []
for a in all_assets:
    if a.metadata_by_key[a.key].get("odd", False):
        odd_assets.append(a)
odd_job = define_asset_job(name="odd_job", selection=odd_assets)

defs = Definitions(
    assets=all_assets,
    jobs=[odd_job]
)

How you create the initial list of all assets could be done with a manually maintained list, or with load_assets_from_package_module or similar type function. If you have SourceAssets or assets from some kinds of integrations (ie airbyte) you may need to add a type check in the loop isinstance(a, AssetsDefinition)

[gnilrets]

Yes! And that works with load_assets_from_package_module because that works recursively. Thanks!