Hi all, Simple question, perhaps, but I'm getting a bit stuck on how to get the result of a source asset when making a dynamic asset. I have the following code:

from dagster import (
    asset,
    AssetsDefinition,
    AssetIn,
    Definitions,
)
import pandas as pd


@asset
def asset_one():
    return pd.DataFrame({"A": [1, 2, 3]})


@asset
def asset_two():
    return pd.DataFrame({"B": [4, 5, 6]})


def asset_factory(source_asset: AssetsDefinition) -> AssetsDefinition:
    source_asset_name = source_asset.asset_key[0][0]
    new_asset_name = source_asset_name + "_to_blob_storage"

    @asset(
        name=new_asset_name,
        group_name="blob",
        non_argument_deps={AssetIn(source_asset_name).key},
    )
    def _asset(context):
        <http://context.log.info|context.log.info>(
            f"This is {new_asset_name} with dependency {source_asset_name}"
        )
        <http://context.log.info|context.log.info>(f"The output dataframe of {source_asset_name} is: ...?")

    return _asset


source_assets = [
    asset_one,
    asset_two,
]
blob_assets = [asset_factory(source) for source in source_assets]

defs = Definitions(assets=[*source_assets, *blob_assets])

I'd like to get the outputs of asset_one and asset_two in my dynamically generated assets. I believe it's something very simple (like 'source_asset.<something>.<something>' but can't for the life of me figure it out. Could anyone lend a hand?

You can use a similar strategy as here to retrieve the asset value off of the

Definitions

object: https://github.com/dagster-io/dagster/discussions/14805

Awesome, this saves me so much work, thanks a lot!