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 user method must be such that the user method is a subtype of the operation defined in the service definition, i.e. respecting input type contravariance and output type covariance.