How do you access the definition metadata for an upstream asset?

[jonathanbhaskar]

If I define metadata in the decorator of an asset, how do I grab that metadata in a downstream asset? For example, in the following code, how do I access the metadata of small_petals in downstream_asset through the context?

@asset(
    metadata={"dataset_name": "iris.small_petals"},
)
def small_petals():
    ...
    pass

@asset(deps=[small_petals])
def downstream_asset(context):
   ...

I understand there is a way to get the latest upstream materialization metadata through the context, but I'm confused about how to access the definition metadata. Any help would be appreciated.

[garethbrickman]

Definition-level metadata is typically meant for showing in the UI in a current context. So it's not automatically persisted on historical materialization events. Nevertheless, you can access it in a downstream asset by importing the asset's definition (defs):

@asset(deps=["usptream"])
def downstream():
    from top_level_module import defs

    upstream_metadata = defs.get_asset_graph().get(AssetKey("upstream")).metadata

[jonathanbhaskar]

Thank you!

[pSpitzner]

Thanks! Very useful answer!

I am just chiming in to suggest adding this answer somewhere clearly visible to the documentation.

I was trying to find out what the table is to which dbt model asset was materialized, and it took me way longer than it should! -> Getting the full metadata of an upstream asset seems like a very general use case that should be easy to find.
(If there is more documentation on this that I could not find, please let me know!)

Minimal working example, project named my_project

# definitions.py

from dagster import Definitions, load_assets_from_modules, with_source_code_references
from dagster_dbt import DbtCliResource

from my_project import dbt_assets, python_assets

all_assets = with_source_code_references(
    load_assets_from_modules([dbt_assets, python_assets])
)


defs = Definitions(
    assets=all_assets,
    resources={
        "dbt": DbtCliResource(project_dir="./dbt"),
    },
)

# dbt_assets.py

import os
from pathlib import Path

from dagster import AssetExecutionContext, Definitions
from dagster_dbt import DbtCliResource, DbtProject, dbt_assets

our_dbt_project = DbtProject(
    project_dir=Path(os.path.abspath(os.environ.get("DBT_PROJECT_DIR"))).resolve(),
    packaged_project_dir=Path(__file__).joinpath("..", "dbt-project").resolve(),
)
our_dbt_project.prepare_if_dev()


@dbt_assets(manifest=our_dbt_project.manifest_path, project=our_dbt_project)
def our_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

# python_assets.py

from dagster import AssetExecutionContext, asset
from dagster_dbt import get_asset_key_for_model

from dagster_haushaltplus.dbt_assets import our_dbt_assets


@asset(
    compute_kind="python",
    deps=[get_asset_key_for_model([our_dbt_assets], "my_dbt_model_name")],
)
def downstream_asset_that_needs_meta(context: AssetExecutionContext) -> None:

    from my_project.definitions import defs

    asset_key = get_asset_key_for_model([our_dbt_assets], "my_dbt_model_name")
    upstream_metadata = defs.get_asset_graph().get(asset_key).metadata
    materialized_table = upstream_metadata.get('dagster/table_name')

    context.log.debug(f"{upstream_metadata=}")
    context.log.debug(f"{materialized_table=}")