nexusrpc.handler._util

module documentation

Undocumented

Function get_start_method_input_and_output_type_annotations Extract input and output type annotations from a start method.
Type Variable ServiceHandlerT Undocumented
def get_start_method_input_and_output_type_annotations(start: Callable[[ServiceHandlerT, StartOperationContext, InputT], OutputT | Awaitable[OutputT]]) -> tuple[type[InputT] | None, type[OutputT] | None]:

Extract input and output type annotations from a start method.

Parameters
start:Callable[[ServiceHandlerT, StartOperationContext, InputT], OutputT | Awaitable[OutputT]]A start method with signature (self, ctx: StartOperationContext, input: I) -> O
Returns
A tuple of (input_type, output_type)
where
  • None means the type annotation is missing or could not be extracted. This is valid when a service definition provides the types.
  • type(None) (NoneType) means the annotation explicitly specified None as the type (e.g., input: None or -> None).

When None is returned for either type, the caller should handle it based on whether a service definition is available:

  • If a service definition is provided, the types from the service definition will be used and type validation against the handler is skipped.
  • If no service definition is provided, an error will be raised downstream when attempting to create an OperationDefinition.
ServiceHandlerT =

Undocumented

Value
TypeVar('ServiceHandlerT')