nexusrpc

package documentation

nexusrpc is a library for building Nexus handlers.

See https://github.com/nexus-rpc and https://github.com/nexus-rpc/api/blob/main/SPEC.md.

Nexus is a synchronous RPC protocol. Arbitrary duration operations are modeled on top of a set of pre-defined synchronous RPCs.

A Nexus caller calls a handler. The handler may respond inline (synchronous response) or return a token referencing the ongoing operation (asynchronous response). The caller can cancel an asynchronous operation using this token. The caller can also specify a callback URL, which the handler uses to deliver the result ofan asynchronous operation when it is ready.

Package	`handler`	Components for implementing Nexus handlers.
Module	`_common`	No module docstring; 2/2 type variables documented
Module	`_serializer`	No module docstring; 1/2 class documented
Module	`_service`	A Nexus service definition is a class with class attributes of type Operation. It must be be decorated with @nexusrpc.service. The decorator validates the Operation attributes.
Module	`_util`	No module docstring; 0/3 constant, 5/7 functions documented

From __init__.py:

Class	`Content`	A container for a map of headers and a byte array of data.
Class	`HandlerErrorType`	Nexus handler error types.
Class	`LazyValue`	A container for a value encoded in an underlying stream.
Class	`Link`	A Link contains a URL and a type that can be used to decode the URL.
Class	`Operation`	Defines a Nexus operation in a Nexus service definition.
Class	`OperationDefinition`	Internal representation of a user's `Operation` definition.
Class	`OperationErrorState`	The state of an operation as described by an `OperationError`.
Class	`OperationState`	Describes the current state of an operation.
Class	`ServiceDefinition`	Internal representation of a user's service definition class.
Exception	`HandlerError`	A Nexus handler error.
Exception	`OperationError`	An error that represents "failed" and "canceled" operation results.
Function	`get_operation`	Return the `nexusrpc.Operation` for the object, or None
Function	`get_service_definition`	Return the `nexusrpc.ServiceDefinition` for the object, or None
Function	`service`	Decorator marking a class as a Nexus service definition.
Function	`set_operation`	Set the `nexusrpc.Operation` for this object.
Type Variable	`InputT`	Operation input type
Type Variable	`OutputT`	Operation output type

def get_operation(obj: Any) -> nexusrpc.Operation[Any, Any] | None: ¶

Return the nexusrpc.Operation for the object, or None

obj should be a decorated operation start method, or a method that takes no arguments and returns an OperationHandler.

def get_service_definition(obj: Any) -> nexusrpc.ServiceDefinition | None: ¶

Return the nexusrpc.ServiceDefinition for the object, or None

@overload
@dataclass_transform(field_specifiers=(Operation))

def service(cls: type[ServiceT]) -> type[ServiceT]:

@overload
@dataclass_transform(field_specifiers=(Operation))

def service(*, name: str | None = None) -> Callable[[type[ServiceT]], type[ServiceT]]:

Decorator marking a class as a Nexus service definition.

The decorator validates the operation definitions in the service definition: that they have the correct type, and that there are no duplicate operation names. The decorator also creates instances of the Operation class for each operation definition.

Example

@nexusrpc.service
class MyNexusService:
    my_op: nexusrpc.Operation[MyInput, MyOutput]
    another_op: nexusrpc.Operation[str, dict]

@nexusrpc.service(name="custom-service-name")
class AnotherService:
    process: nexusrpc.Operation[ProcessInput, ProcessOutput]

def set_operation(obj: Any, operation: nexusrpc.Operation[Any, Any]): ¶

Set the nexusrpc.Operation for this object.

obj should be an operation start method.

InputT = ¶

Operation input type

Value

TypeVar('InputT',
        contravariant=True)

OutputT = ¶

Operation output type

Value

TypeVar('OutputT',
        covariant=True)