nexusrpc.handler._operation_handler

module documentation

Undocumented

Class SyncOperationHandler An OperationHandler that is limited to responding synchronously.
Function bind_operation_handler_methods_to_service Give each operation handler method of a service handler class its own operation.
Function collect_operation_handler_factories_by_method_name Collect operation handler methods from a user service handler class.
Function service_definition_from_operation_handler_methods Create a service definition from operation handler factory methods.
Function validate_operation_handler_methods Validate operation handler methods against a service definition.
Function _wrap_factory Undocumented
Function _wrap_inherited_method Undocumented
def bind_operation_handler_methods_to_service(service_cls: type[ServiceHandlerT], service_definition: ServiceDefinition):

Give each operation handler method of a service handler class its own operation.

The operation of each method reports service_definition as its service. The operation gets its name and types from the service definition. When service_cls inherits a method from a parent class, this function replaces the method on service_cls with a wrapper. Thus, the method of the parent class keeps its own operation.

Validate the methods against the service definition before you call this function.

def collect_operation_handler_factories_by_method_name(user_service_cls: type[ServiceHandlerT], service: ServiceDefinition | None) -> dict[str, Callable[[ServiceHandlerT], OperationHandler[Any, Any]]]:

Collect operation handler methods from a user service handler class.

def service_definition_from_operation_handler_methods(service_name: str, user_methods: dict[str, Callable[[ServiceHandlerT], OperationHandler[Any, Any]]]) -> ServiceDefinition:

Create a service definition from operation handler factory methods.

In general, users should have access to, or define, a service definition, and validate their service handler against it by passing the service definition to the @nexusrpc.handler.service_handler decorator. This function is used when that is not the case.

def validate_operation_handler_methods(service_cls: type[ServiceHandlerT], operation_handler_factories_by_method_name: dict[str, Callable[[ServiceHandlerT], OperationHandler[Any, Any]]], service_definition: ServiceDefinition):

Validate operation handler methods against a service definition.

For every operation in service_definition:

  1. There must be a method in user_methods whose method name matches the method name from the service definition.
  2. The input and output types of the handler method must exactly match the types declared in the service definition.
def _wrap_factory(factory: Callable[[Any], OperationHandler[Any, Any]], op: Operation[Any, Any]) -> Callable[[Any], OperationHandler[Any, Any]]:

Undocumented

def _wrap_inherited_method(method: Callable[..., Any]) -> Callable[..., Any]:

Undocumented