IO Manager that Depends on Resource

example.py

from typing import Any
from dagster import ConfigurableResource, ConfigurableIOManager, InputContext, OutputContext, asset, Definitions, ResourceDependency, EnvVar
from pydantic import Field

# https://docs.dagster.io/concepts/resources#resources-that-depend-on-other-resources

class myResource(ConfigurableResource):
    username: str = Field(description="the username")
    password: str = Field(description="the password")

class myIOManager(ConfigurableIOManager):
    credentials: ResourceDependency[myResource]
    account: str = Field(description="the account to use")

    def handle_output(self, context: OutputContext, obj: Any) -> None:
        context.log.info(f"Storing output using {self.account} with {self.credentials.username}")

    def load_input(self, context: InputContext) -> Any:
        context.log.info(f"Loading inputs with {self.account} and {self.credentials.username}")
        

@asset
def upstream(): ...

@asset
def downstream(upstream): ...

defs = Definitions(
    assets=[upstream, downstream],
    resources={
        "io_manager": myIOManager(
            credentials=myResource(
                username=EnvVar("USERNAME"), 
                password=EnvVar("PASSWORD")
            ),
            account="us-west2"
        )
    }
)