nanover.core.nanover_server module

class nanover.core.nanover_server.NanoverServer(*, address: str, port: int, max_workers=128)

Bases: GrpcServer

A base for NanoVer gRPC servers. Sets up a gRPC server, and automatically attaches a CommandService and StateService enabling the running of arbitrary commands and synchronisation of state.

close()

Stops the server.

Inheritors of this class should override this method with routines to stop services that are running.

property commands: Dict[str, CommandRegistration]

Gets the commands available on this server.

Returns:

The commands, consisting of their names, callback and default parameters.

copy_state() → Dict[str, object]

Return a deep copy of the current state.

lock_state() → ContextManager[Dict[str, object]]

Context manager for reading the current state while delaying any changes to it.

register_command(name: str, callback: Callable[[...], Dict[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any] | Iterable[None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]] | Mapping[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]]] | None] | Callable[[], Dict[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any] | Iterable[None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]] | Mapping[str, None | str | int | float | bool | Iterable[Any] | Mapping[str, Any]]] | None], default_arguments: Dict | None = None)

Registers a command with the CommandService running on this server.

Parameters:

  • name – Name of the command to register

  • callback – Method to be called whenever the given command name is run by a client.

  • default_arguments – A description of the arguments of the callback and their default values.

Raises:

ValueError – Raised when a command with the same name already exists.

setup_services()

Sets up the services, including the CommandService.

unregister_command(name)

Deletes a command from this service.

Parameters:

name – Name of the command to delete

update_locks(access_token: object = None, acquire: Dict[str, float] | None = None, release: Set[str] | None = None)

Attempts to acquire and release locks on keys in the shared key/value store. If any of the locks cannot be acquired, none of them will be.

update_state(access_token: object, change: DictionaryChange)

Attempts an atomic update of the shared key/value store. If any key cannot be updates, no change will be made.