define_asset_job accept AssetKey and AssetsDefinition selections

[sryza]

Summary & Motivation

Enable passing a list of AssetKeys or AssetsDefinitions to the selection argument of define_asset_job.

How I Tested These Changes

bk

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

2 Ignored Deployments

Name	Status	Preview	Comments	Updated
dagit-storybook	⬜️ Ignored (Inspect)			Jan 9, 2023 at 5:51PM (UTC)
dagster	⬜️ Ignored (Inspect)			Jan 9, 2023 at 5:51PM (UTC)

[NicolasPA]

Thank you for this! It will feel much more natural to be able to directly pass a bunch of AssetsDefinition to an asset selection parameter.
I think this changed should be applied to every asset selection parameter, for example @asset_sensor and @multi_asset_sensor (related to #11567). Below before/after examples of the code simplification it leads to:

from dagster import asset, define_asset_job, AssetSelection, with_resources, multi_asset_sensor, RunRequest


@asset
def asset1():
    return [1, 2, 3]


@asset
def asset2(asset1):
    return asset1 + [4]


@asset
def asset3():
    return 1


# Instead of:
job1 = define_asset_job(name="job1", selection=AssetSelection.assets(asset1))
# Directly pass AssetsDefinition
job1 = define_asset_job(name="job1", selection=asset1)

# Instead of:
job2 = define_asset_job(name="job2", selection=AssetSelection.assets(asset1, asset2))
# pass List[AssetsDefinition]
job2 = define_asset_job(name="job2", selection=[asset1, asset2])

assets_with_resources = with_resources(
    [asset1, asset2],
    resource_defs={"foo": my_resource()}
)

# Instead of:
job3 = define_asset_job(name="job3", selection=AssetSelection.assets(*assets_with_resources))
# pass Sequence[AssetsDefinition]
job3 = define_asset_job(name="job3", selection=assets_with_resources)

# Instead of:
job4 = define_asset_job(name="job4", selection=AssetSelection.assets(asset3))
# pass AssetsDefinition
job4 = define_asset_job(name="job4", selection=asset3)


# Instead of:
@multi_asset_sensor(
    asset_selection=AssetSelection.assets(*assets_with_resources),
    job=job4,
)
def asset_1_and_2_sensor(context):
    return RunRequest()


# pass Sequence[AssetsDefinition]
@multi_asset_sensor(
    asset_selection=assets_with_resources,
    job=job4,
)
def asset_1_and_2_sensor(context):
    return RunRequest()

[OwenKephart]

Happy to approve as is, but thoughts on creating a CoercibleToAssetSelection type (similar to the CoercibleToAssetKey type)? This would make it easier to apply this same logic to other asset selection parameters.

[sryza]

Happy to approve as is, but thoughts on creating a CoercibleToAssetSelection type (similar to the CoercibleToAssetKey type)? This would make it easier to apply this same logic to other asset selection parameters.

I'm going to merge this in its present state, but yes, I think we should do this as soon as we add the same parameter to a second function/class.