[pythonic-resources] ConfigurableResource.from_resource_context_cm does not handle nested resources

[fahadkh]

Dagster version

dagster, version 1.6.14

What's the issue?

When working with nested Pythonic resources, the from_resource_context_cm does not properly instantiate an instance with nested resources defined.

For example:

    class UpstreamResource(ConfigurableResource):
        a: str

    class DownstreamResource(ConfigurableResource):
        upstream: ResourceDependency[UpstreamResource]
        b: str

    with DownstreamResource.from_resource_context_cm(
        build_init_resource_context(
            config={"b": "world"},
            resources={
                "upstream": UpstreamResource(a="hello")
            }
        )
    ) as resource:
        assert resource.upstream.a == "hello"

errors with:

>           assert resource.upstream.a == "hello"
E           AttributeError: 'NoneType' object has no attribute 'a'

What did you expect to happen?

The instantiated resource should have a properly instantiated nested resource (i.e. the test case above should pass)

How to reproduce?

Full test case:

def test_nested_resource_from_context():

    class UpstreamResource(ConfigurableResource):
        a: str

    class DownstreamResource(ConfigurableResource):
        upstream: ResourceDependency[UpstreamResource]
        b: str

    with DownstreamResource.from_resource_context_cm(
        build_init_resource_context(
            config={"b": "world"},
            resources={
                "upstream": UpstreamResource(a="hello")
            }
        )
    ) as resource:
        assert resource.upstream.a == "hello"

Deployment type

Other Docker-based deployment

Deployment details

Running on kubernetes

Additional information

May be related to https://github.com/dagster-io/dagster/pull/19422

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.

[rni-HMC]

Not sure if this is the same root cause as this issue:

from dagster import ConfigurableResource, op, job, ResourceDependency

# Dagger
class UpstreamResource(ConfigurableResource):
    param: str

class DownstreamResource(ConfigurableResource):
    upstream: ResourceDependency[UpstreamResource]

    def do_something(self):
        return f"upstream param is: {self.upstream.param}"

# User job.py
@op()
def my_op(outer: DownstreamResource):
    return outer.do_something()

@job()
def my_job():
    my_op()

# User defs.py
from dagster import Definitions

defs = Definitions(
    jobs=[my_job],
    resources={
        "outer": DownstreamResource(upstream=UpstreamResource.configure_at_launch())
    },
)

# User job.py
run_config = {
    "resources": {"outer": {"config": {"upstream": {"param": "configured at launch"}}}}
}

# Execute the job with the provided run configuration
result = defs.get_job_def("my_job").execute_in_process(run_config=run_config)
print(result.success)

Results in this error: dagster._check.CheckError: Invariant failed. Description: Any partially configured, nested resources must be provided to Definitions object: {'upstream': None}