Is there any way to access

op_config

within an IOManager? E.g.

@op(config_schema = {"uuid": str})
def op1(context) -> int:
    return 1

@op(config_schema = {"uuid": str})
def op2(context, a: int):
    return a * 2

@job(resource_defs={"io_manager": my_io_manager})
def my_job():
    op2(op1())

@io_manager
def my_io_manager():
    return MyIOManager()

class MyIOManager(IOManager):

    def handle_output(self, context, obj):
        # I want to access the uuid here to form the filepath to save the object

    def load_input(self, context):
        # I want to access the uuid here to form the filepath to load the object

Yep, you can do something like this:

class MyIOManager(IOManager):
    def handle_output(self, context, obj):
        context.step_context.op_config

From the docs, we can pass static per-output Metadata to the IO Manager

@op(out=Out(metadata={"schema": "some_schema", "table": "some_table"}))
def op_1():
    """Return a Pandas DataFrame"""


@op(out=Out(metadata={"schema": "other_schema", "table": "other_table"}))
def op_2(_input_dataframe):
    """Return a Pandas DataFrame"""

class MyIOManager(IOManager):
    def handle_output(self, context, obj):
        table_name = context.metadata["table"]
        schema = context.metadata["schema"]
        write_dataframe_to_table(name=table_name, schema=schema, dataframe=obj)

    def load_input(self, context):
        table_name = context.upstream_output.metadata["table"]
        schema = context.upstream_output.metadata["schema"]
        return read_dataframe_from_table(name=table_name, schema=schema)


@io_manager
def my_io_manager(_):
    return MyIOManager()

My question: Is there any way to specify this metadata as configuration? E.g. the “schema” and “table” can be decided beforehand similarly to the field is the op’s

op_config

?

@op(out=Out(metadata={"schema": str, "table": str}))
def op_1():
    """Return a Pandas DataFrame"""