And the second question is that: For the same function, when the

DagsterRunStatus.SUCCESS

shows in the log, the

context.dagster_run.asset_selection

is None. Why is that

If a run targets all the assets in the job, then the asset_selection field won't get filled out. If you want to find all the materializations that were planned as part of a run, you can do

materializations_planned = context.instance.get_records_for_run(
            run_id=run_id, of_type=DagsterEventType.ASSET_MATERIALIZATION_PLANNED
        ).records
        asset_keys = set(cast(AssetKey, record.asset_key) for record in materializations_planned)

Hi Sandy, but in our case, only a single Asset is included in the Run

Does the job have more than a single asset?

No, every job we have only has a single asset

In that case, the run is targeting all the assets in the job, so the asset_selection field won't get filled out The above snippet should still work though for finding the asset targeted by the job