If I'm running the materialize error metrics step, how do I pass in an arbitrary electrothermal model materialization result as an input? The pattern @yuhan mentioned below would work if error metrics materialization is running right after a electrothermal materialization, but I want to be able to materialize error metrics on any electrothermal model.

@asset
def upstream():
    return MaterializeResult(metadata={"table_name": "foo", "file_path": "bar"})

@asset(deps=[upstream])
def downstream_asset(context):
    materialization = context.instance.get_latest_materialization_event(upstream.key).asset_materialization
    metadata = materialization.metadata
    assert metadata["table_name"].value == "foo"
    assert metadata["file_path"].value == "bar"

I could pass in a config.electrothermal_model_id to error_metrics that would be used if the upstream electrothermal_model input is None, but that seems hacky. I'm also not sure if dagster would allow that

In the provided code snippet, you are using the

get_latest_materialization_event

method to retrieve the metadata from the most recent materialization of the

upstream

asset. This method is part of the Dagster instance API and allows you to access the metadata of the latest materialization event for a given asset key. If you want to materialize error metrics based on any electrothermal model materialization result, you can indeed pass a configuration parameter to the

error_metrics

asset that specifies which electrothermal model to use. This is not necessarily hacky; it's a common pattern to use configuration to parameterize the behavior of assets based on runtime conditions. Here's a conceptual example based on the documentation:

python @asset def electrothermal_model(context):     # ... your logic to materialize the electrothermal model ...     return MaterializeResult(metadata={"model_id": "some_id", "other_data": "value"})  @asset(deps=[electrothermal_model]) def error_metrics(context):     # If an explicit model ID is provided in the config, use it to fetch the specific materialization     model_id = context.op_config.get("electrothermal_model_id")     if model_id:         # Fetch the specific materialization event based on the provided model_id         # You would need to implement `get_materialization_event_by_model_id` to suit your needs         materialization_event = get_materialization_event_by_model_id(context.instance, model_id)     else:         # Otherwise, use the latest materialization of the electrothermal_model         materialization_event = context.instance.get_latest_materialization_event(electrothermal_model.key).asset_materialization      # Access the metadata from the materialization event     metadata = materialization_event.metadata     # ... your logic to compute error metrics based on the metadata ...

In this example,

get_materialization_event_by_model_id

is a hypothetical function you would implement to fetch a materialization event based on a specific model ID. The

context.op_config

is used to access the configuration passed to the asset at runtime. Please note that the exact implementation details, such as how to fetch a materialization event by a specific ID, depend on your application's logic and are not detailed in the documentation. The concept of passing configuration parameters to assets to control their behavior at runtime is, however, a standard approach in Dagster.