https://dagster.io/ logo
Join Slack
Powered by
hello! noob question here. what's the best way to ...
# ask-community
m
hello! noob question here. what's the best way to configure my custom io manager (below) to write files to the same directory that the default 
io_manager
does?
Copy code
class JsonIOManager(IOManager):
    def _get_path(self, context) -> str:
        if context.has_partition_key:
            return "/".join(context.asset_key.path + [context.asset_partition_key])
        else:
            return "/".join(context.asset_key.path)

    def handle_output(self, context, obj):
        if obj is None:
            return

        with open(self._get_path(context), "w") as file:
            file.write(json.dumps(obj))

    def load_input(self, context):
        return json.load(open(self._get_path(context.upstream_output)))
should i be inheriting from 
ConfigurableIOManager
instead? i'm a bit confused when i should use one over the other? also is there an example for how to inherit from 
UPathIOManager
the api looked a bit overwhelming blob smile sweat
y
the default io manager defaults the base dir to 
context.instance.storage_directory()
, so the easiest path could be to call that instance method in your 
_get_path
❤️ 1
🙏 1
the difference between IOManager and ConfigurableIOManager is whether you’d want to use Pythonic/Pydantic configs. the latter supports it. here’s the doc with an example: https://docs.dagster.io/concepts/resources#defining-pythonic-io-managers
m
awesome! that's exactly what I was looking for. Thank you
❤️ 1
weird that didn't seem to work. i get an error
Copy code
dagster._core.errors.DagsterExecutionHandleOutputError: Error occurred while handling output "result" of step "test_raw_json"::

AttributeError: 'OutputContext' object has no attribute 'instance'
and i just checked my installation and I am on 
dagster==1.3.2
y
sorry my bad!
Copy code
class MyIOManager(ConfigurableIOManager):
    root_path: str

    def _get_path(self, asset_key: AssetKey) -> str:
        return self.root_path + "/".join(asset_key.path)

    def handle_output(self, context: OutputContext, obj):
        # write_csv(self._get_path(context.asset_key), obj)
        print(self.get_resource_context().instance.storage_directory())

    def load_input(self, context: InputContext):
        return read_csv(self._get_path(context.asset_key))
the context is resource context, so when you’re inside an io manager’s handle_output or load_input which is at runtime, you can access it via 
self.get_resource_context().instance.storage_directory()
m
hmm i'm still getting an error 
AttributeError: 'OutputContext' object has no attribute 'get_resource_context'
here's what I have after running dir on the context object
Copy code
[
    "__annotations__",
    "__class__",
    "__del__",
    "__delattr__",
    "__dict__",
    "__dir__",
    "__doc__",
    "__enter__",
    "__eq__",
    "__exit__",
    "__format__",
    "__ge__",
    "__getattribute__",
    "__gt__",
    "__hash__",
    "__init__",
    "__init_subclass__",
    "__le__",
    "__lt__",
    "__module__",
    "__ne__",
    "__new__",
    "__reduce__",
    "__reduce_ex__",
    "__repr__",
    "__setattr__",
    "__sizeof__",
    "__str__",
    "__subclasshook__",
    "__weakref__",
    "_asset_info",
    "_config",
    "_dagster_type",
    "_events",
    "_log",
    "_mapping_key",
    "_metadata",
    "_name",
    "_op_def",
    "_partition_key",
    "_pipeline_name",
    "_resource_config",
    "_resources",
    "_resources_cm",
    "_run_id",
    "_step_context",
    "_step_key",
    "_user_events",
    "_user_generated_metadata",
    "_version",
    "_warn_on_step_context_use",
    "add_output_metadata",
    "asset_info",
    "asset_key",
    "asset_partition_key",
    "asset_partition_key_range",
    "asset_partition_keys",
    "asset_partitions_def",
    "asset_partitions_time_window",
    "config",
    "consume_events",
    "consume_logged_metadata",
    "dagster_type",
    "get_asset_identifier",
    "get_asset_output_identifier",
    "get_identifier",
    "get_logged_events",
    "get_logged_metadata",
    "get_output_identifier",
    "get_run_scoped_output_identifier",
    "has_asset_key",
    "has_asset_partitions",
    "has_partition_key",
    "log",
    "log_event",
    "mapping_key",
    "metadata",
    "name",
    "op_def",
    "partition_key",
    "pipeline_name",
    "resource_config",
    "resources",
    "run_id",
    "step_context",
    "step_key",
    "version",
]
y
did you extend from ConfigurableIOManager ?
oh it’s gonna be 
self
rather than 
context
❤️ 1
m
Ah beautiful! works great. Glad I asked this one confused me a bit.
❤️ 1
65 Views
Open in Slack
PreviousNext
Powered by