nexusrpc.handler._operation_handler

module documentation

Undocumented

Class SyncOperationHandler An OperationHandler that is limited to responding synchronously.
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.
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.