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.

clear_locks()

Forces the release all locks on all keys in the shared key/value store.

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.