def METHOD_NAME(self, context, step_output_data, dagster_type):
    step_output_handle = step_output_data.step_output_handle
    manager = context.get_io_manager(step_output_handle)
    manager_key = context.execution_plan.get_manager_key(step_output_handle, self.job_def)
    res = manager.load_input(
        context.for_input_manager(
            name=None,
            config=None,
            metadata=None,
            dagster_type=dagster_type,
            source_handle=step_output_handle,
            resource_config=context.resolved_run_config.resources[manager_key].config,
            resources=build_resources_for_manager(manager_key, context),
        )
    )
    return res