Hoping I'm just missing something in the docs, but I'm struggling to figure out how to launch Asset materializations from an op or job, at least in a way that allows me to really customize which Assets get launched at runtime. Please excuse the rough snippet below, but something along the lines of...

@asset
def render_asset_1(context):
    # ... do something...
    return something

assets = [render_asset_1, ...]

@op
def determine_assets_to_launch(context):
    for asset in assets:
        # ... some logic on which assets are currently needed...
        if criteria:
            asset() # Run the asset


@job
def dynamic_asset_job():
    # the job that will ultimately run on a schedule.
    determine_assets_to_launch()

My reason for something like this is I would like to have a series of explicitly defined assets that can take full advantage of dagit and adhoc materializations, while also being able to dynamically execute the assets through a single job. Something that doesn't seem possible with just

define_asset_job

based on an

AssetSelection

group which will always execute every Asset, nor can it be done through something like just an op with

AssetMaterialization

, which doesn't render the Asset in dagit for adhoc launches.

Check this link

Gotcha! And so then the AssetSelection will be evaluated prior to each job run to determine which Assets it will be running that time?

Here is the doc, is very limited but is what we have: https://docs.dagster.io/_apidocs/assets#dagster.AssetSelection

Ah okay! Last question - in a scenario like this, is there a way to get hold of basic Asset event details such as last materialization time? I know how to do so within an op/asset using

context.instance.get...()

, but how would I do something like this outside of that?

If I understand well, you can use a sensor and get all the that information

In response to the last question about getting basic asset event details, it's easiest to do it inside of an op/asset/sensor where you have access to the instance and can fetch the latest materialization per asset, etc. Outside of that, you'd probably have to get your dagster instance and then fetch the information you need.

with DagsterInstance.get() as instance:
   ...