I found this: <https://github.com/dagster-io/dagst...
# ask-community
I found this: https://github.com/dagster-io/dagster/issues/5699 But the docs for multi_asset_sensors describe different funcitonality.
I updated my asset_sensor to make unique (more unique) run_key and added tags to to the run_request. Appears to function now.
🌈 1
Glad you got it to work--feel free to post again in #dagster-support if any other issues come up
Thank you for your help. I am having difficulty extracting metadata from asset_materialization in the sensor. In my case, my asset returns a dict of values on materialization. In the sensor, I parse those values and pass to Job in RunConfig.
def ma_multi_download_sensor(context):
# returns a list of materialization event records for a specified asset key.
all_asset_a_events = context.materialization_records_for_key(
asset_key=AssetKey("aid_ma_multi_samp"), limit=5
Currently, the only meta data value I am able to get to is the 'Path' to storage. The file in storage contains the key value pairs I am looking for.
for event in all_asset_a_events:
asset_event_metadata = {
key: val.value
for key, val in event.asset_materialization.metadata.items()
{'path': '/tmp/tmp3qg0hm9m/storage/aid_ma_multi_samp'}
Thank you
Found my issue passing data from asset to sensor. Needed to set value and metadata in yielded object.