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`	Undocumented
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	`Handler`	A Nexus handler manages a collection of Nexus service handlers.
Class	`MiddlewareSafeOperationHandler`	An `OperationHandler` where `start` and `cancel` can be awaited by an async runtime. It can produce a result synchronously by returning `StartOperationResultSync` or asynchronously by returning `StartOperationResultAsync`...
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	`OperationHandlerMiddleware`	Middleware for operation handlers.
Class	`OperationTaskCancellation`	Indicates whether a a Nexus task has been cancelled during a sync operation or before an async operation has returned a token.
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	`operation_handler`	Decorator marking an operation handler factory method in a service handler class.
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 operation_handler(method: OperationHandlerFactoryT) -> OperationHandlerFactoryT:

@overload

def operation_handler(*, name: str | None = None) -> Callable[[OperationHandlerFactoryT], OperationHandlerFactoryT]:

Decorator marking an operation handler factory method in a service handler class.

An operation handler factory method is a method that takes no arguments other than self and returns an OperationHandler instance.

Parameters
method:`OperationHandlerFactoryT \| None`	The method to decorate.
name:`str \| None`	Optional name for the operation. If not provided, the method name will be used.

Returns
`OperationHandlerFactoryT \| Callable[[OperationHandlerFactoryT], OperationHandlerFactoryT]`	Undocumented

@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] \| None`	The service handler class to decorate.
service:`type[Any] \| None`	The service definition that the service handler implements.
name:`str \| None`	Optional 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}"