context

numerous.tasks.context

Task execution context — provides access to controller and inputs.

get_task_controller()

Get the task controller for the current task execution.

Source code in numerous/tasks/context.py

def get_task_controller() -> TaskController:
    """Get the task controller for the current task execution."""
    controller = _current_controller.get()
    if controller is not None:
        return controller

    instance_id = os.getenv("NUMEROUS_TASK_INSTANCE_ID")
    if instance_id is None:
        msg = (
            "No task controller available. "
            "Must be called within a task execution context, "
            "or NUMEROUS_TASK_INSTANCE_ID must be set (platform mode)."
        )
        raise RuntimeError(msg)

    controller = _create_platform_controller(instance_id)
    _current_controller.set(controller)
    return controller

get_task_inputs()

Get the inputs for the current task execution.

Source code in numerous/tasks/context.py

def get_task_inputs() -> dict[str, Any]:
    """Get the inputs for the current task execution."""
    inputs = _current_inputs.get()
    if inputs is not None:
        return inputs

    input_data = os.getenv("TASK_DATA_INPUT")
    inputs = deserialize_task_inputs(input_data) if input_data else {}

    _current_inputs.set(inputs)
    return inputs