task

numerous.tasks.task

Task management API - Functional Interface.

create_task_definition(func, name=None, app_id=None)

Create a task definition from a function.

Source code in numerous/tasks/task.py

def create_task_definition(
    func: Callable[..., Any],
    name: Optional[str] = None,
    app_id: Optional[str] = None,
) -> TaskDefinition:
    """Create a task definition from a function."""
    return TaskDefinition(
        id=str(uuid.uuid4()),
        name=name or func.__name__,
        func=func,
        app_id=app_id,
    )

create_task_instance(task_def, inputs, workload=TaskWorkload.LOCAL)

Create a new task instance state.

Source code in numerous/tasks/task.py

def create_task_instance(
    task_def: TaskDefinition,
    inputs: dict[str, Any],
    workload: TaskWorkload = TaskWorkload.LOCAL,
) -> TaskInstanceState:
    """Create a new task instance state."""
    return TaskInstanceState(
        id=str(uuid.uuid4()),
        task_id=task_def.id,
        status=TaskStatus.PENDING,
        progress=0.0,
        inputs=inputs,
        workload=workload,
        created_at=datetime.now().astimezone(),
    )

get_task_definition(task_id)

Get a task definition by ID.

Source code in numerous/tasks/task.py

def get_task_definition(task_id: str) -> Optional[TaskDefinition]:
    """Get a task definition by ID."""
    return _store.get_task_definition(task_id)

get_task_instance(instance_id)

Get a task instance by ID.

Source code in numerous/tasks/task.py

def get_task_instance(instance_id: str) -> Optional[TaskInstanceState]:
    """Get a task instance by ID."""
    return _store.get_task_instance(instance_id)

list_task_definitions()

List all registered task definitions.

Source code in numerous/tasks/task.py

def list_task_definitions() -> list[TaskDefinition]:
    """List all registered task definitions."""
    return _store.list_task_definitions()

list_task_instances(task_id=None)

List all task instances, optionally filtered by task_id.

Source code in numerous/tasks/task.py

def list_task_instances(task_id: Optional[str] = None) -> list[TaskInstanceState]:
    """List all task instances, optionally filtered by task_id."""
    return _store.list_task_instances(task_id)

register_instance(state)

Register a task instance.

Source code in numerous/tasks/task.py

def register_instance(state: TaskInstanceState) -> TaskInstanceState:
    """Register a task instance."""
    return _store.register_instance(state)

register_task(task_def)

Register a task definition globally.

Source code in numerous/tasks/task.py

def register_task(task_def: TaskDefinition) -> TaskDefinition:
    """Register a task definition globally."""
    return _store.register_task(task_def)

request_stop(state)

Request the task to stop.

Source code in numerous/tasks/task.py

def request_stop(state: TaskInstanceState) -> TaskInstanceState:
    """Request the task to stop."""
    if state.controller:
        state.controller.request_stop()
    return state

run_task(task_def, inputs, block=False, workload=TaskWorkload.LOCAL)

High-level function to create and execute a task.

This composes the lower-level functions into a useful operation.

Source code in numerous/tasks/task.py

def run_task(
    task_def: TaskDefinition,
    inputs: dict[str, Any],
    block: bool = False,  # noqa: FBT001,FBT002
    workload: TaskWorkload = TaskWorkload.LOCAL,
) -> TaskInstanceState:
    """
    High-level function to create and execute a task.

    This composes the lower-level functions into a useful operation.
    """
    # Create instance
    state = create_task_instance(task_def, inputs, workload)
    register_instance(state)

    # Get executor
    executor = _get_executor()

    # Submit for execution
    future = submit_task_execution(task_def, state, executor)

    # Optionally block
    if block:
        future.result()

    return state

set_progress(state, progress)

Set task progress, delegating to controller if available.

Source code in numerous/tasks/task.py

def set_progress(state: TaskInstanceState, progress: float) -> TaskInstanceState:
    """Set task progress, delegating to controller if available."""
    if state.controller:
        state.controller.set_progress(progress)
    return state

set_status(state, status)

Set task status, delegating to controller if available.

Source code in numerous/tasks/task.py

def set_status(state: TaskInstanceState, status: TaskStatus) -> TaskInstanceState:
    """Set task status, delegating to controller if available."""
    if state.controller:
        state.controller.set_status(status)
    return state

stop_task_instance(instance_id)

Stop a running task instance.

Source code in numerous/tasks/task.py

def stop_task_instance(instance_id: str) -> Optional[TaskInstanceState]:
    """Stop a running task instance."""
    state = get_task_instance(instance_id)
    if state:
        return request_stop(state)
    return None

submit_task_execution(task_def, state, executor)

Submit task for async execution and return future.

Source code in numerous/tasks/task.py

def submit_task_execution(
    task_def: TaskDefinition,
    state: TaskInstanceState,
    executor: TaskExecutor,
) -> Future[Any]:
    """Submit task for async execution and return future."""
    future = executor.submit(task_def, state)
    state.future = future
    return future

task(func=None, *, name=None, app_id=None)

task(
    func: Callable[..., R],
    *,
    name: Optional[str] = None,
    app_id: Optional[str] = None
) -> Callable[..., TaskInstanceState]

task(
    func: None = None,
    *,
    name: Optional[str] = None,
    app_id: Optional[str] = None
) -> Callable[
    [Callable[..., R]], Callable[..., TaskInstanceState]
]

Convert a function into a task.

Returns a new function that when called, creates and runs a task instance.

Usage

@task def my_task(x: int) -> int: return x + 1

@task(name="custom_name") def another_task(x: int) -> int: return x * 2

Calling the decorated function runs the task

instance = my_task(5)

Source code in numerous/tasks/task.py

def task(
    func: Optional[Callable[..., R]] = None,
    *,
    name: Optional[str] = None,
    app_id: Optional[str] = None,
) -> Union[
    Callable[..., TaskInstanceState],
    Callable[[Callable[..., R]], Callable[..., TaskInstanceState]],
]:
    """
    Convert a function into a task.

    Returns a new function that when called, creates and runs a task instance.

    Usage:
        @task
        def my_task(x: int) -> int:
            return x + 1

        @task(name="custom_name")
        def another_task(x: int) -> int:
            return x * 2

        # Calling the decorated function runs the task
        instance = my_task(5)
    """

    def decorator(f: Callable[..., R]) -> Callable[..., TaskInstanceState]:
        # Inside a platform task instance just return the original function
        if not _is_platform_task_instance():
            task_def = create_task_definition(f, name=name, app_id=app_id)
            register_task(task_def)
        else:
            task_def = None

        @wraps(f)
        def wrapper(*args: Any, **kwargs: Any) -> TaskInstanceState:  # noqa: ANN401
            # Inside a platform task instance just run the function directly.
            if _is_platform_task_instance():
                return f(*args, **kwargs)  # type: ignore[return-value]

            # Convert args/kwargs to inputs dict
            sig = inspect.signature(f)
            params = list(sig.parameters.keys())

            # Build inputs
            inputs = {}
            for i, arg in enumerate(args):
                if i < len(params):
                    inputs[params[i]] = arg
            inputs.update(kwargs)

            # Run the task
            if task_def is None:
                msg = "task_def is not available"
                raise RuntimeError(msg)
            return run_task(task_def, inputs)

        if task_def is not None:
            wrapper.task_def = task_def  # type: ignore[attr-defined]
            wrapper.list_instances = lambda: list_task_instances(task_def.id)  # type: ignore[attr-defined]

        return wrapper

    if func is None:
        # Called with arguments: @task(name="...")
        return decorator
    # Called without arguments: @task
    return decorator(func)

wait_for_completion(state)

Wait for task completion and return result.

Source code in numerous/tasks/task.py

def wait_for_completion(state: TaskInstanceState) -> Any:  # noqa: ANN401
    """Wait for task completion and return result."""
    if state.future:
        return state.future.result()
    return state.result