nectarengine.rpc module

class nectarengine.rpc.RPC(url: str | None = None, user: str | None = None, password: str | None = None, **kwargs: Any)

Bases: object

This class allows to call API methods synchronously, without callbacks.

It logs warnings and errors.

Usage:

from nectarengine.rpc import RPC
rpc = RPC()
print(rpc.getLatestBlockInfo(endpoint="blockchain"))
get_request_id() int

Get request id.

request_send(endpoint: str, payload: bytes) str
rpcexec(endpoint: str, payload: List[Dict[str, Any]]) Any

Execute a call by sending the payload.

Parameters:

payload (json) – Payload data

Raises:
  • ValueError – if the server does not respond in proper JSON format

  • RPCError – if the server returns an error

version_string_to_int(network_version: str) int
exception nectarengine.rpc.RPCError

Bases: Exception

RPCError Exception.

exception nectarengine.rpc.RPCErrorDoRetry

Bases: Exception

RPCErrorDoRetry Exception.

class nectarengine.rpc.SessionInstance

Bases: object

Singelton for the Session Instance

instance: Client | None = None
exception nectarengine.rpc.UnauthorizedError

Bases: Exception

UnauthorizedError Exception.

nectarengine.rpc.get_endpoint_name(*args: Any, **kwargs: Any) str
nectarengine.rpc.set_session_instance(instance: Client) None

Set session instance

nectarengine.rpc.shared_session_instance() Client

Get session instance