I'd like to be able to define my IO_Manager on an asset using a partition key like so.

@asset(
    resource_defs={
        "io_manager": ConfigurablePickledObjectGCSIOManager,
    },
    partitions_def=MultiPartitionsDefinition(
        {
            "date": DailyPartitionsDefinition(start_date="2023-07-10"),
            "customer": StaticPartitionsDefinition(customer_list),
        }
    ),
    context_kwargs={
        "bucket": Config(context.partition_key["customer"]).buckets["raw_data"],
        "prefix": "zendesk/chats",
    },
)
def zendesk_chats(context):
    ...

Of course the

@asset

doesn't have access to the

context

object, how can I get around this? Open to other approaches. I see how this could be an anti-pattern. Assets are supposed to read/write to the same storage.

Hi Liam. Unfortunately it's not possible to configure at job runtime the IO manager that will be used. I'd say the most straightforward way to write to a bucket based on partition would be to create your IO manager that determines the bucket to write to using the input/output context's partition key attribute, within the

load_input

or

handle_output

calls.