ray.serve.schema.TaskProcessorAdapter#

class ray.serve.schema.TaskProcessorAdapter(*args, **kwargs)[source]#

Bases: ABC

Abstract base class for task processing adapters.

Subclasses can support different combinations of sync and async operations. Use supports_async_capability() to check if a specific async operation is supported.

PublicAPI (alpha): This API is in alpha and may change before becoming stable.

abstract initialize(consumer_concurrency: int = 5)[source]#

Initialize the task processor.

abstract register_task_handle(func: Callable, name: str | None = None)[source]#

Register a function as a task handler.

Parameters:

  • func – The function to register as a task handler.

  • name – Custom name for the task.

abstract enqueue_task_sync(task_name: str, args: Any | None = None, kwargs: Any | None = None, **options) → TaskResult[source]#

Enqueue a task for execution synchronously.

Parameters:

  • task_name – Name of the registered task to execute.

  • args – Positional arguments to pass to the task function.

  • kwargs – Keyword arguments to pass to the task function.

  • **options – Additional adapter-specific options for task execution.

Returns:

Object containing task ID, status, and other metadata.

Return type:

TaskResult

abstract get_task_status_sync(task_id: str) → TaskResult[source]#

Retrieve the current status of a task synchronously.

Parameters:

task_id – Unique identifier of the task to query.

Returns:

Object containing current task status, result, and other metadata.

Return type:

TaskResult

abstract start_consumer(**kwargs)[source]#

Start the task consumer/worker process.

abstract stop_consumer(timeout: float = 10.0)[source]#

Stop the task consumer gracefully.

Parameters:

timeout – Maximum time in seconds to wait for the consumer to stop.

abstract cancel_task_sync(task_id: str)[source]#

Cancel a task synchronously.

Parameters:

task_id – Unique identifier of the task to cancel.

abstract get_metrics_sync() → Dict[str, Any][source]#

Get metrics synchronously.

Returns:

Adapter-specific metrics data.

Return type:

Dict[str, Any]

abstract health_check_sync() → List[Dict][source]#

Perform health check synchronously.

Returns:

Health status information for workers/components.

Return type:

List[Dict]

async enqueue_task_async(task_name: str, args: Any | None = None, kwargs: Any | None = None, **options) → TaskResult[source]#

Enqueue a task asynchronously.

Parameters:

  • task_name – Name of the registered task to execute.

  • args – Positional arguments to pass to the task function.

  • kwargs – Keyword arguments to pass to the task function.

  • **options – Additional adapter-specific options for task execution.

Returns:

Object containing task ID, status, and other metadata.

Return type:

TaskResult

Raises:

NotImplementedError – If subclass didn’t implement enqueue_task_async function

async get_task_status_async(task_id: str) → TaskResult[source]#

Get task status asynchronously.

Parameters:

task_id – Unique identifier of the task to query.

Returns:

Object containing current task status, result, and other metadata.

Return type:

TaskResult

Raises:

NotImplementedError – If subclass didn’t implement get_task_status_async function

async cancel_task_async(task_id: str)[source]#

Cancel a task.

Parameters:

task_id – Unique identifier of the task to cancel.

Raises:

NotImplementedError – If subclass didn’t implement cancel_task_async function

async get_metrics_async() → Dict[str, Any][source]#

Get metrics asynchronously.

Returns:

Adapter-specific metrics data.

Return type:

Dict[str, Any]

Raises:

NotImplementedError – If subclass didn’t implement get_metrics_async function

async health_check_async() → List[Dict][source]#

Perform health check asynchronously.

Returns:

Health status information for workers/components.

Return type:

List[Dict]

Raises:

NotImplementedError – If subclass didn’t implement health_check_async function