Hello Dagster team and users, I have been struggling to extend the functionality of the Google Sheets I/O Manager example. I would like to add the ability for the I/O manager to name and place the generated spreadsheet differently depending on the metadata associated with the asset's output. Inspired by the pattern described in the documentation, I tried using the following return statement at the end of my asset's definition:

return Output(
    value=df,
    metadata={
        "table_title": "Order Data",
        "destination_folder_id": "12345ABCD"
    }
)

When I tried to access the

context.metadata

object within the

handle_output

function of the I/O manager that I specified to handle the asset's output, however, I received an empty object. I can confirm that the I/O manager received the asset's output because I can access the asset's dataframe in the

handle_output

function. What happened? Is there a simpler way to have the I/O manager perform the stated goal?

yeah, unfortunately this is a missing piece from the API that the dagster team is aware. It was recently discussed here: https://dagster.slack.com/archives/C01U954MEER/p1682604438916379 You can fetch the metadata using the event logs.. Below is a a helper function that gets called in the

handle_output

and receives the context as a parameter. It is not pretty but it gets the job done. I hope that we can have an easier way to do this in the near future.

def _get_destination_info(self, context: OutputContext) -> tuple:
        metadata_values = {}
        _folder_id = None
        _table_title = None


        events = context.step_context.instance.all_logs(
            run_id=context.run_id, of_type=DagsterEventType.STEP_OUTPUT
        )
        for e in events:
            event_specific_data = e.dagster_event.event_specific_data
            if event_specific_data.step_output_handle.output_name == context.name:
                # for dynamic outputs, we need to consider the mapping key
                if event_specific_data.step_output_handle.mapping_key is None:
                    metadata_values = event_specific_data.metadata
                    break
                elif event_specific_data.step_output_handle.mapping_key == context.mapping_key:
                    metadata_values = event_specific_data.metadata
                    break

        for metadata, value in metadata_values.items():
            if metadata == "table_title":
                _table_title = value.value
            if metadata == "destination_folder_id":
                _folder_id = value.value
        if not _table_title or not _folder_id:
            raise ValueError('Metadata "table_title" and/or "destination_folder_id" not found.')
        return _folder_id, _table_title

Thanks for the insight @André Augusto