Hi team! I noticed that whenever I use

resource_defs

in a

@job

, any resource being used in the underlying

@op

s also has to be defined in such job's

resource_defs

because the job wouldn't look for resource keys defined in my project's definition. For example:

@op
def my_op(aircall_resource: Aircall):
   ...

@job
def my_job():
  my_op()

defs = Definitions(jobs=[my_job], resources={"aircall_resource":Aircall()}

This would work fine. But,

from dagster import mem_io_manager

@op(out=Out(io_managaer_key="mem_manager"))
def my_op(aircall_resource: Aircall):
   ...

@job(resource_defs={"mem_manager": mem_io_manager})
def my_job():
  my_op()

defs = Definitions(jobs=[my_job], resources={"aircall_resource":Aircall()}

this would say

aircall_resource

not defined in defs. I would then have to do this:

from dagster import mem_io_manager

@op(out=Out(io_managaer_key="mem_manager"))
def my_op(aircall_resource: Aircall):
   ...

@job(resource_defs={"mem_manager": mem_io_manager, "aircall_resource": Aircall()})
def my_job():
  my_op()

defs = Definitions(jobs=[my_job], resources={"aircall_resource":Aircall()}) # aircall_resource would be used by other @ops, so i'd have to duplicate its definition

Do you know if this is expected? If so, what would be the best way to define these Out io managers without rewriting resource defs in the jobs?

Hi Dennis, right now I believe this isn’t possible - the suggested workaround would be what you posted in the second comment. I think a structure like that should work - what error are you seeing?