I think I may be running into an issue with the

PythonObjectDagsterType

and it’s associated

loader

argument. The

loader

was created using the

@dagster_type_loader

decorator and requests a resource via

required_resource_keys={"file_downloader"}

. So put together I have something like this:

@dagster_type_loader(..., required_resource_keys={"file_downloader"})
def type_loader(context, config) -> MyType:
  s3_downloader: S3Downloader = context.resources.file_downloader
  return MyType(path=s3_downloader.download(...))

DagitConfigurableMyType = PythonObjectDagsterType(
    python_type=MyType, loader=type_loader
)

Now I can use this as an input to a graph just fine - and configure it via dagit as an input to a graph. The

file_downloader

resource is resolved and brought in as desired:

@graph(
    ...
    ins={"path": In(DagitConfigurableMyType)},
)
def intake_graph(in_path: MyType):
    ...

However doing this almost works:

@graph(
    ...
    ins={"path": In(list[DagitConfigurableMyType])},
)
def intake_graphs(in_paths: list[MyType]):
    ...

Except that it complains about not being able to resolve the

file_downloader

resource. I’m assuming this is somehow related to how resources are resolved. If I manually add the resource to the first op in the graph it will actually resolve the resource for me. However, since the

dagster_type_loader

does get respected even when nested within a

list

, it seems like the resource should get resolved as well

hi @fahad! thanks for the report, looks like you're right. I filed an issue for this here: https://github.com/dagster-io/dagster/issues/9161, and in the meantime your solution of explicitly adding the resource to the op as well is probably the best workaround

Ah thanks for creating the issue!