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, check for its outcome, or fetch its current state. The caller can also specify a callback URL, which the handler uses to deliver the result of an 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 OperationInfo Information about an operation.
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)