Bernardo Cortez
01/06/2022, 3:59 PM@asset_sensor
updated?prha
01/06/2022, 4:07 PMasset_sensor
implementation is a wrapper around this event query:
event_records = context.instance.get_event_records(
EventRecordsFilter(
event_type=DagsterEventType.ASSET_MATERIALIZATION,
asset_key=self._asset_key,
after_cursor=after_cursor,
),
ascending=False,
limit=1,
)
event_record = event_records[0]
... # body of asset sensor is evaluated here
context.update_cursor(str(event_record.storage_id))
The cursor is updated with the record id of the most recent asset materialization event.
Reference: https://github.com/dagster-io/dagster/blob/master/python_modules/dagster/dagster/core/definitions/sensor_definition.py#L570