How to get the metadata or tags for the latest materialization of an asset?

[jamiedemaria]

Versions of this question that come up:

• How can I get an asset's (asset_1) metadata in another asset (asset_2) or in an op?
• In a sensor, how can I get the metadata from the latest materialization of a particular asset?
• When materializing an asset, how can I get the metadata from the previous materialization of the asset?

Additionally, how can I get the tags from the latest materialization of an asset?

[jamiedemaria]

The context object that's passed to functions decorated by @asset, @op, and @sensor contains an instance attribute that can query the Dagster event log.

If you're running code outside of an asset, op, or sensor, you can access the instance by using DagsterInstance.get(): #12808.

Example of how to access metadata from a historical materialization of one asset from inside another asset

Unpartitioned

@asset
def asset1():
    return Output(value=..., metadata={"num_rows": 5})

@asset  # or op
def asset_1_num_rows(context):
    instance = context.instance
    materialization = instance.get_latest_materialization_event(AssetKey(["asset1"])).asset_materialization
    return materialization.metadata["num_rows"]

Partitioned

from dagster import (
    asset,
    StaticPartitionsDefinition,
    EventRecordsFilter,
    AssetKey,
    Output,
    DagsterEventType,
)


@asset(partitions_def=StaticPartitionsDefinition(["a", "b"]))
def asset1():
    return Output(value=..., metadata={"num_rows": 5})


@asset  # or op
def asset1_a_num_rows(context):
    instance = context.instance
    materialization = instance.get_event_records(
        event_records_filter=EventRecordsFilter(
            event_type=DagsterEventType.ASSET_MATERIALIZATION,
            asset_key=AssetKey(["asset1"]),
            asset_partitions=["a"],
        ),
        limit=1,
    )[0].asset_materialization
    return materialization.metadata["num_rows"]

Note that you can also get previous materializations metadata of an asset in the implementation of the same asset

Unpartitioned example

@asset
def asset1(context):
    instance = context.instance
    materialization = instance.get_latest_materialization_event(AssetKey(["asset1"])).asset_materialization
    previous_num_rows = materialization.metadata["num_rows"]

    return Output(value=..., metadata={"num_rows": 5})

[sryza]

To add to this, if you want to see the tags for the run that materialized the asset, you can do the following:

materialization_event = instance.get_latest_materialization_event(AssetKey(["asset_1"]))
run_id = materialization_event.run_id
run_tags = instance.get_run_by_id(run_id).tags

[alex-orlovskyi]

Hi, @sryza, @jamiedemaria.
Consider having a run with dynamic partition X created. How can we query asset_1's metadata from asset_2 for current partition only?
I see, that there's get_latest_materialization_events method available, but wondering what's the most efficient way to filter by partition.

[sryza]

@alex-orlovskyi I added code snippet to the example above with a partitioned asset

[alex-orlovskyi]

@sryza, amazing! Thank you) That's what i've been looking for.

[marcilj]

Hey @jamiedemaria, I tried all of this and it didn't worked.

Than I focus on the terminology and I think I found out why this didn't work for me.

When you say get the metadata and the tags you are talking about Output Metadata and Run tags right?

Is it possible to also retrieve the metadata information about the asset, the tags/op_tags defined in the asset decorator or the owners ?

I'm trying to create my own alerts sensor and triage alerts based on the teams which they belong, and the only available field I can think of are tags, metadata or owners, but those didn't seem available with the context.

@asset(
    op_tags={
        "kind": "python",
        "severity": "critical",
        "owner_tag": "DE",
    },
    owners=["team:DE"],
    metadata={
        "owner": "my_email_address@email.com",
        "priority": "high",
    },
) 
def my asset():
  return 1

Are any of the elements listed in the asset above code is retrievable from a sensor after a run failure or a run success?

Thank you

[don1uppa]

How do you get the returned object from the asset we are looking for. This won't run: event_records_filter=EventRecordsFilter(
event_type=DagsterEventType.HANDLED_OUTPUT,
asset_key=AssetKey([asset_key]),
asset_partitions=[context.partition_key],
)