nexusrpc.handler

package documentation

Components for implementing Nexus handlers.

Server/worker authors will use this module to create the top-level Nexus handlers responsible for dispatching requests to Nexus operations.

Nexus service/operation authors will use this module to implement operation handler methods within service handler classes.

Module _common Undocumented
Module _core This module contains Handler classes. A Handler manages a collection of Nexus service handlers. It receives and responds to incoming Nexus requests, dispatching to the corresponding operation handler.
Module _decorators No module docstring; 0/2 type variable, 1/1 function documented
Module _operation_handler No module docstring; 3/3 functions, 1/1 class documented
Module _syncio No module docstring; 1/1 class documented
Module _util No module docstring; 0/1 type variable, 1/1 function documented

From __init__.py:

Class CancelOperationContext Context for the cancel method.
Class FetchOperationInfoContext Context for the fetch_info method.
Class FetchOperationResultContext Context for the fetch_result method.
Class Handler A Nexus handler manages a collection of Nexus service handlers.
Class OperationContext Context for the execution of the requested operation method.
Class OperationHandler Base class for an operation handler in a Nexus service implementation.
Class StartOperationContext Context for the start method.
Class StartOperationResultAsync A value returned by the start method of a nexus operation handler indicating that the operation is responding asynchronously.
Class StartOperationResultSync A result returned synchronously by the start method of a nexus operation handler.
Function service_handler Decorator that marks a class as a Nexus service handler.
Function sync_operation Decorator marking a method as the start method for a synchronous operation.
@overload
def service_handler(cls: type[ServiceHandlerT]) -> type[ServiceHandlerT]:
@overload
def service_handler(*, service: type[Any] | None = None) -> Callable[[type[ServiceHandlerT]], type[ServiceHandlerT]]:
@overload
def service_handler(*, name: str) -> Callable[[type[ServiceHandlerT]], type[ServiceHandlerT]]:

Decorator that marks a class as a Nexus service handler.

A service handler is a class that implements the Nexus service by providing operation handler implementations for all operations in the service.

The class should implement Nexus operation handlers as methods decorated with operation handler decorators such as @nexusrpc.handler.operation_handler.

Example

from nexusrpc.handler import service_handler, sync_operation

@service_handler(service=MyService)
class MyServiceHandler:
    @sync_operation
    async def my_operation(
        self, ctx: StartOperationContext, input: MyInput
    ) -> MyOutput:
        return MyOutput(processed=input.data)
Parameters
cls:type[ServiceHandlerT] | NoneThe service handler class to decorate.
service:type[Any] | NoneThe service definition that the service handler implements.
name:str | NoneOptional name to use for the service, if a service definition is not provided. service and name are mutually exclusive. If neither is provided, the class name will be used.
Returns
type[ServiceHandlerT] | Callable[[type[ServiceHandlerT]], type[ServiceHandlerT]]Undocumented
@overload
def sync_operation(start: Callable[[ServiceHandlerT, StartOperationContext, InputT], Awaitable[OutputT]]) -> Callable[[ServiceHandlerT, StartOperationContext, InputT], Awaitable[OutputT]]:
@overload
def sync_operation(start: Callable[[ServiceHandlerT, StartOperationContext, InputT], OutputT]) -> Callable[[ServiceHandlerT, StartOperationContext, InputT], OutputT]:
@overload
def sync_operation(*, name: str | None = None) -> Callable[[Callable[[ServiceHandlerT, StartOperationContext, InputT], Any]], Callable[[ServiceHandlerT, StartOperationContext, InputT], Any]]:

Decorator marking a method as the start method for a synchronous operation.

Example

import httpx
from nexusrpc.handler import service_handler, sync_operation

@service_handler
class MyServiceHandler:
    @sync_operation
    async def process_data(
        self, ctx: StartOperationContext, input: str
    ) -> str:
        # You can use asynchronous I/O libraries
        async with httpx.AsyncClient() as client:
            response = await client.get("https://api.example.com/data")

        data = response.json()
        return f"Processed: {data}"