Pythonic resource depending on non-pythonic resource I'm gradually migrating existing non-pythonic resources to pythonic resources. In general, I have resources that depend on other resources. Is it possible to have a pythonic resource depend on a non-python resource? To illustrate, I'll use a modified example from the documentation (https://docs.dagster.io/concepts/resources#resources-which-depend-on-other-resources)

from dagster import Definitions, ConfigurableResource, resource

# Non-python resource
class Credentials:
   def __init__(self, username, password):
       self.username = username
       self.password = password

@resource(config_schema={"username": str, "password": str})
def credentials_resource(context: dagster.InitResourceContext) -> Credentials:
    return Credentials(username=context.resource_config["username"], password=context.resource_config["password"])

# Pythonic resource
class FileStoreBucket(ConfigurableResource):
    credentials: ??? # How can I express a dependency on a non-python resource?
    region: str

    def write(self, data: str):
        get_filestore_client(
            username=self.credentials.username,
            password=self.credentials.password,
            region=self.region,
        ).write(data)

defs = Definitions(
    assets=[my_asset],
    resources={
        "bucket": FileStoreBucket(
            credentials=credentials_resource(
                username="my_user", password="my_password"
            ),
            region="us-east-1",
        ),
    },
)

If I understand you question correctly, you might want to use

setup_for_execution

function: https://docs.dagster.io/concepts/resources#managing-state-in-resources. Then you could express

credentials

as a `PrivateAttr`:

class FileStoreBucket(ConfigurableResource):
  _credentials: Credentials = PrivateAttr()
  
  def setup_for_execution(self, context) -> None:
    self._credentials = Credentials()

@Timur Iskhakov I believe that's exactly what I need. Thank you for that tip. I've tried to make your suggest work as follows

from dagster import Definitions, ConfigurableResource, resource

# Non-python resource
class Credentials:
   def __init__(self, username, password):
       print(f"{username}:{password}")
       self.username = username
       self.password = password

@resource(config_schema={"username": str, "password": str})
def credentials_resource(context: dagster.InitResourceContext) -> Credentials:
    return Credentials(username=context.resource_config["username"], password=context.resource_config["password"])

# Pythonic resource
class FileStoreBucket(ConfigurableResource):
    region: str
    _credentials: Credentials = pydantic.PrivateAttr()
    
    def setup_for_execution(self, context):
        print(type(context.resources.credentials))
        self._credentials = context.resources.credentials

    def write(self, data: str):
        print(data)

@dagster.op(required_resource_keys={"credentials", "bucket"})
def the_step(context):
    bucket = context.resources.bucket
    bucket.write("data")

@dagster.job(
    resource_defs={
        "bucket": FileStoreBucket(
            region="us-east-1",
        ),
        "credentials": credentials_resource
    }
)
def the_job():
    the_step()

the_job.execute_in_process(run_config={'resources': {'credentials': {'config': {'password': 'the_password', 'username': 'the_username'}}}})

But I'm getting the error

Unknown resource `credentials`. Specify `credentials` as a required resource on the compute / config function that accessed it.

I guess it makes sense that the

context

passed to

setup_for_execution

needs to know that it's dependent on

credentials

, but I don't know how to specify that dependency.

@ben are you able to take a look at this pythonic resource-related question?

Hi Stefan, the following should work:

from dagster import ResourceDependency, ConfigurableResource, resource, Definitions
import dagster

# Non-python resource
class Credentials:
   def __init__(self, username, password):
       print(f"{username}:{password}")
       self.username = username
       self.password = password

@resource(config_schema={"username": str, "password": str})
def credentials_resource(context: dagster.InitResourceContext) -> Credentials:
    return Credentials(username=context.resource_config["username"], password=context.resource_config["password"])

# Pythonic resource
class FileStoreBucket(ConfigurableResource):
    region: str
    credentials: ResourceDependency[Credentials]
    
    def write(self, data: str):
        print(data)

@dagster.op(required_resource_keys={"credentials", "bucket"})
def the_step(context):
    bucket = context.resources.bucket
    bucket.write("data")

@dagster.job(
)
def the_job():
    the_step()

defs = Definitions(
    jobs=[the_job],
    resources={
        "bucket": FileStoreBucket(
            region="us-east-1",
            credentials=credentials_resource
        ),
        "credentials": credentials_resource
    }
)
defs.get_job_def("the_job").execute_in_process(run_config={'resources': {'credentials': {'config': {'password': 'the_password', 'username': 'the_username'}}}})

Yes @ben! Good man!