Spencer Nelson
01/24/2023, 1:33 AM@asset
decorator’s io_manager_key
parameter supposed to work? I can’t seem to get it to recognize a custom IOManager even with a very minimal case (code in thread)class MyIOManager(IOManager):
def _get_path(self, context) -> str:
return "/".join(context.asset_key.path)
def handle_output(self, context, obj):
pass
def load_input(self, context):
return 1
@io_manager
def my_io_manager(init_context):
return MyIOManager()
@asset(io_manager_key="my_io_manager")
def my_asset(context):
pass
When I try to load this, I get an error:dagster._core.errors.DagsterInvalidDefinitionError: io manager with key 'my_io_manager' required by output 'result' of op 'my_asset'' was not provided. Please provide a <class 'dagster._core.storage.io_manager.IOManagerDefinition'> to key 'my_io_manager', or change the required key to one of the following keys which points to an <class 'dagster._core.storage.io_manager.IOManagerDefinition'>: ['io_manager']
io_manager_def
works, for example:
@asset(io_manager_def=my_io_manager)
def my_asset(context):
pass
Zach
01/24/2023, 1:37 AMSpencer Nelson
01/24/2023, 1:40 AMdefs = Definitions(
assets=[my_asset],
resources={
"my_io_manager": my_io_manager
},
)
at the end of the file has no effectfrom dagster import asset, IOManager, io_manager, Definitions
class MyIOManager(IOManager):
def _get_path(self, context) -> str:
return "/".join(context.asset_key.path)
def handle_output(self, context, obj):
pass
def load_input(self, context):
return 1
@io_manager
def my_io_manager(init_context):
return MyIOManager()
@asset(io_manager_key="my_io_manager")
def my_asset(context):
pass
defs = Definitions(
assets=[my_asset],
resources={
"my_io_manager": my_io_manager
},
)
Alec Koumjian
02/06/2023, 4:39 PM@job
, it will not properly load the resource definitions. However, if you use define_asset_job
, it loads in the global resource definitions just fine.