Hello there. I’m trying to build reusable IO_Managers, and I have a hard time finding how to linked ressources together. In the example I found I can do something like this

@io_manager(required_resource_keys={"s3_bucket", "s3"})
def common_bucket_s3_io_manager(init_context):
    """
    A version of the s3_csv_io_manager that gets its bucket from another resource.
    """
    return s3_csv_io_manager(
        build_init_resource_context(
            config={"s3_bucket": init_context.resources.s3_bucket},
            resources={"s3": init_context.resources.s3},
        )
    )

Where

s3_bucket

and

s3

are required by this new ressource so, If I configured all the right ressources in my

project/resources/__init__.py

file I’ll be able to make that work. Is there a way to specify which ressources to use instead of hardcoding them? I’m asking because I need to write to 2 different buckets that required 2 differents AWS credentials in different jobs. If

s3

and

s3_bucket

is taken but the first one, how do I make the second IO_MANAGER work? I know I could create a new io manager let’s say called

my_new_other_common_bucket_s3_io_manager

like this

@io_manager(required_resource_keys={"my_new_s3_bucket", "my_new_s3_client_ressource"})
def my_new_other_common_bucket_s3_io_manager(init_context):
    """
    A version of the s3_csv_io_manager that gets its bucket from another resource.
    """
    return s3_csv_io_manager(
        build_init_resource_context(
            config={"s3_bucket": init_context.resources.my_new_s3_bucket},
            resources={"s3": init_context.resources.my_new_s3_client_ressource},
        )
    )

But this feels wrong… Is there a better way?

hey @Jacob Marcil a couple months ago we release a new way to write resources/io managers that will solve this problem for you! This page will show you how to author them https://docs.dagster.io/concepts/resources the important part for you is that the resources/io managers you write will now be classes, and you will pass resource dependencies to each class. So to make a simplified version of your example, you could do

class CommonBucketS3IOManager(ConfigurableIOManager):
    s3: S3Resource 
    
    ....

defs = Defintions(
    assets=[...],
    resources={
        "io_mgr_1": CommonBucketS3IOManager(s3=S3Resource(<some_config>)),
        "io_mgr_2": CommonBucketS3IOManager(s3=S3Resource(<other_config>))
    }
)

Oh interesting. Thank you I already add the IoManager defined as a class but was using function in order to use that

@io_manager

decorator and return a class object out of it. I’ll try that 🙂

@jamie what's the benefit to inheriting from ConfigurableIOManager vs IOManager?

configureable io manager will allow you to set up the resource dependencies and add other configuration, like above. IOManager is just an interface that enforces the handle output and load input functions