Is there a simple way to get the current

data_version

of an upstream asset, perhaps in the

context

somewhere?

there's not a way to do this through purely public API methods at the moment (so it'll be a bit ugly, and things might change in the future), but one way to do this would be

from dagster._core.definitions.data_version import (
    extract_data_version_from_entry,
)

@asset
def foo(context, upstream_asset):
    upstream_entry = context.get_step_execution_context().get_input_asset_record(AssetKey("upstream_asset")).event_log_entry
    upstream_version = extract_data_version_from_entry(upstream_entry)

This is great, thanks. I figured out on my own this:

latest_record = context.instance.get_latest_data_version_record(
        AssetKey("asset_key")
    )

But using the execution context seems better since it will guarantee we are using the event log entry that corresponds with the version loaded in the downstream asset rather than the latest.