nanover.core.commands module

class nanover.core.commands.CommandMessageHandler(send_message: SendMessage, *, command_service: CommandService | None = None)

Bases: object

Handle the message protocol for remote commands by handling incoming messages and emitting outgoing messages. This is essentially (part of) an RPC implementation.

handle_message(message)

Handle an incoming command message, which may be a request to register a new command, a request to invoke a command, etc.

Note: for now this will block when fulfilling a command registered remotely.

register_command(name: str, callback: Callable[[...], dict | None], default_arguments: dict | None = None) → None

Register a local callback that can be invoked by a remote party.

request_command(name: str, arguments: dict | None = None) → Future

Request the invocation of a remote callback.

class nanover.core.commands.CommandService(add_list_command=True)

Bases: object

Implementation of the Command service, enabling services to register arbitrary commands which are run as callbacks.

property commands: dict[str, CommandRegistration]

Gets a copy of the commands that have been registered, as CommandRegistration, including their names, default arguments and registered callback.

Returns:

A copy of the dictionary of commands that have been registered.

register_command(name: str, callback: Callable[[...], dict | None], default_arguments: dict | None = None)

Registers a command with this service

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 dictionary of the arguments of the callback and their default values.

Raises:

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

run_command(name: str, arguments: dict) → Future

unregister_command(name)

Deletes a command from this service.

Parameters:

name – Name of the command to delete

class nanover.core.commands.SendMessage(*args, **kwargs)

Bases: Protocol