How to get the data version of an upstream asset, while computing a downstream asset

[sryza]

Given an ObservableSourceAsset and a downstream asset, how do I get the dataversion associated with the source asset in downstream asset?

[sryza]

from dagster import asset, observable_source_asset, DataVersion


@observable_source_asset
def asset1():
    return DataVersion("5")


@asset(non_argument_deps={"asset1"})
def asset2(context):
    latest_materialization_event = context.get_step_execution_context().get_input_asset_record(asset1.key).event_log_entry
    upstream_data_version = latest_materialization_event.tags.get("dagster/data_version")

This should also work if the upstream asset is a regular materializable asset instead of an observable source asset.

[amardeep]

I tried this, and it doesn't work as latest_materialization_event is None (when running via dagster dev -f ...) after observing asset1 in dagit and trying to materialize asset2.

On the other hand, if I follow https://discuss.dagster.io/t/10381103/is-there-a-simple-way-to-get-the-current-data-version-of-an- and use following, it works:

latest_materialization_event = context.get_step_execution_context().get_input_asset_record(asset1.key).event_log_entry
upstream_data_version = latest_materialization_event.tags.get("dagster/data_version")

[sryza]

Oops - you are right @amardeep - updating my answer to avoid misleading people with an incorrect example

[lfrkncht]

thanks @amardeep and @sryza for your solution. unfortunately it did not work for me, the following did the job for me

@observable_source_asset
def asset1():
    return DataVersion("5")

@asset(deps=[asset1])
def asset2(context):
    upstream_data_version = context.get_step_execution_context().input_asset_records[asset1.key].data_version