Hello, I have two quick questions about asset materializations. What is the best way of checking the last time a given asset was materialized? How can I check if an asset (or any one of a group of assets) is currently being materialized?

hey @Rohin do you need to do these checks in code? or are you just looking for where that info is in the UI?

I would like to do these checks in code.

so to do that you’ll need to query the dagster instance via the context. you can use

@op
def materialization_event_times(context):
   context.instance.get_latest_materialization_event(asset_key=AssetKey("foo)).timestamp

to get the time when an asset was materialized. Checking if an asset is currently being materialized is a bit more complicated and requires importing some non-public apis

from dagster._core.storage.pipeline_run import IN_PROGRESS_RUN_STATUSES, RunsFilter
@op
def current_materializations(context):
    event = context.instance.get_event_records(event_records_filter=EventRecordsFilter(event_type=DagsterEventType.ASSET_MATERIALIZATION_PLANNED, asset_key=AssetKey("foo")), limit=1)
    in_progress = context.instance.get_runs(
                        filters=RunsFilter(
                            run_ids=[
                                event[0].event_log_entry.run_id
                            ],
                            statuses=IN_PROGRESS_RUN_STATUSES,
                        )
                    )

    # in_progress is a bool of where the materialization run is currently executing

Great, that is very helpful, thank you!