How to: fetch asset materializations from the instance database

[sryza]

Sometimes you want to write a Python script that fetches asset materializations from the instance database. Or to query the instance database in a sensor.

[sryza]

Note that the APIs mentioned here are not guaranteed to be stable across releases

If you're inside a sensor, schedule, or op, you can get the instance from the context:

instance = context.instance

If not, you can get the instance using the approach described here: #7427.

Once you have the instance, you can query the events:

from dagster.core.storage.event_log.base import EventRecordsFilter
from dagster.core.events import DagsterEventType

for record in instance.get_event_records(
    EventRecordsFilter(event_type=DagsterEventType.ASSET_MATERIALIZATION), limit=5
):
    materialization = record.asset_materialization
    asset_key = materialization.asset_key
    metadata = materialization.metadata

If you only want events for a particular asset key, you can specify the asset_key property on your EventRecordsFilter.