https://dagster.io/ logo
#ask-community
Title
# ask-community
c

Chris Histe

07/04/2023, 8:40 PM
Hello support, is there a way to access in an IO Manager the currently running partition in a setup similar to this example: https://docs.dagster.io/concepts/partitions-schedules-sensors/partitions#defining-a-job-with-static-partitions
context.partition_key
would be nice on
OutputContext
but it seems only set for @assets
It’s fine it’s not possible but then is there a way to detect my job is partitioned so I could a throw an error saying partitioned jobs are not handled
c

claire

07/06/2023, 8:02 PM
Hi Chris, I'm running the following code example on the most recent dagster version and
context.partition_key
is working as expected:
Copy code
class CustomIOManager(ConfigurableIOManager):
    def handle_output(self, context: OutputContext, obj: Any) -> None:
        print("my partition key", context.partition_key)
        return None

    def load_input(self, context: InputContext) -> Any:
        ...

@io_manager
def custom_io_manager():
    return CustomIOManager()


@static_partitioned_config(partition_keys=CONTINENTS)
def continent_config(partition_key: str):
    return {"ops": {"continent_op": {"config": {"continent_name": partition_key}}}}


class ContinentOpConfig(Config):
    continent_name: str


@op
def continent_op(context, config: ContinentOpConfig):
    <http://context.log.info|context.log.info>(config.continent_name)


@job(config=continent_config, resource_defs={"io_manager": custom_io_manager})
def continent_job():
    continent_op()
What version of dagster are you running? Does the error still occur with the code snippet above?
c

Chris Histe

07/07/2023, 6:43 PM
Hi Claire, sorry, I cannot reproduce my bug ago from 3 days ago. You are correct it’s indeed set. All good. Sorry for the waste of time 😞
c

claire

07/07/2023, 6:44 PM
glad its working!