| | |
- _CsvRpcServerInterface(builtins.object)
-
- CsvRpcServer(_CsvRpcServerInterface, JsonRpcServer)
- PyCsvRpcServer(_CsvRpcServerInterface, PyJsonRpcServer)
- builtins.object
-
- AsyncServer
- BinaryJsonFormat
-
- BinaryJsonRpcRequestFormat
- BinaryJsonRpcResponseFormat
- Error
- JsonRpcServer
-
- BinaryJsonRpc
- CsvRpcServer(_CsvRpcServerInterface, JsonRpcServer)
- PyJsonRpcServer
-
- PyCsvRpcServer(_CsvRpcServerInterface, PyJsonRpcServer)
- PyRequest
- Request
- RequestHeaders
- Response
- ResponseHeaders
- enum.IntEnum(builtins.int, enum.Enum)
-
- FORMATS
class AsyncServer(builtins.object) |
| |
AsyncServer(host: str, port: int)
This class implements a basic asynchronous TCP server.
|
| |
Methods defined here:
- __init__(self, host: str, port: int)
- Initialize self. See help(type(self)) for accurate signature.
- async handle_client(reader: asyncio.streams.StreamReader, writer: asyncio.streams.StreamWriter) -> None
- This method gets data and parses it from socket.
- async start(self) -> None
- This method start the asynchronous TCP server.
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
|
class BinaryJsonFormat(builtins.object) |
| |
BinaryJsonFormat(encoding: str, get_content: Callable = None)
This class implements the binary JSON format parsing.
|
| |
Methods defined here:
- __init__(self, encoding: str, get_content: Callable = None)
- Initialize self. See help(type(self)) for accurate signature.
- dict_block(self, size_length: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- dict_dump(self, value: Dict[str, ~Json], data: bytearray) -> None
- This method dumps a Json like dict into binary format.
- dumps_data(self, data: ~Json, data_dump: bytearray) -> None
- This method encodes Json like data in binary format.
- duration(self) -> datetime.timedelta
- This method returns timedelta from binary data.
- duration_dump(self, value: datetime.timedelta) -> bytes
- This function dumps a timedelta (duration) to data bytes.
- float_dump(self, data: float) -> bytes
- This function dumps the float value.
- int_block(self, size: int, signed: bool = True) -> int
- This method transfrom the next block value into an integer.
- int_to_block(self, value: int, identifier: bool = True) -> Tuple[bytes, int]
- This method returns a block for a integer value.
- list_block(self, size_length: int) -> List[~Json]
- This method parses a Json like list from binary data.
- list_dump(self, value: List[~Json], data: bytearray) -> None
- This method dumps a Json like list into binary format.
- read(self, number: int) -> bytes
- This method reads `number` of bytes.
- read_dict(self, size: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- read_float4(self) -> float
- This function parses next 4 bytes as float.
- read_float8(self) -> float
- This function parses next 8 bytes as float.
- read_list(self, size: int) -> List[~Json]
- This method parses a Json like list from binary data.
- read_next_element(self) -> ~Json
- This function reads the next Json like element from binary data.
- read_str(self, size: int) -> str
- This method returns strings value from the binary data.
- string_block(self, size_length: int) -> str
- This method transfrom the next block value into an integer.
- string_to_block(self, value: str) -> bytes
- This method returns a block for a string value.
- unix_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to unix data bytes.
- unix_timestamp(self) -> datetime.datetime
- This method returns datetime for a unix timestamp from binary data.
- windows_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to windows data bytes.
>>> BinaryJsonFormat.windows_datetime_to_dump(None, datetime(2025, 6, 22))
b'\x1e\x01\xdb\xe2\xf7\xd5>p\x00'
>>>
- windows_timestamp(self) -> datetime.datetime
- This method returns datetime for a windows timestamp from binary data.
- write_block_size(self, element: ~Json, base: int, bytemap: Dict[int, bytes]) -> Tuple[bytes, bool]
- This method returns a block size and a boolean
to represent the usage of base.
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
|
class BinaryJsonRpc(JsonRpcServer) |
| |
This class implements the binary JSON RPC protocols.
|
| |
- Method resolution order:
- BinaryJsonRpc
- JsonRpcServer
- builtins.object
Class methods defined here:
- handle_request_data(binary: bytes, reader: Callable = None) -> Optional[str] from builtins.type
- This method parses binary JSON from data to perform RPC calls.
>>> BinaryJsonRpc.register_function(dir)
>>> BinaryJsonRpc.register_function(int)
>>> BinaryJsonRpc.handle_request_data(b'\x80Cdir\x00')
>>> BinaryJsonRpc.handle_request_data(b'\x90\x03\x01Lbuiltins.int\x81A1')
bytearray(b'\x00\x03\x01\x03\x01')
>>>
Data and other attributes defined here:
- __annotations__ = {}
Class methods inherited from JsonRpcServer:
- call(method: Callable, json: Dict[str, Any]) -> Any from builtins.type
- This function calls the python function.
- check_request(json: Dict[str, Any], id: Union[str, int, NoneType]) -> Optional[Dict[str, Union[str, int, JsonRpcExtended.Error]]] from builtins.type
- This function checks the JSON RPC received requests.
>>> JsonRpcServer.register_function(dir, 'test')
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.check_request({}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0'}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'test', 'params': None}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'builtins.dir'}, 1)
>>>
- execute_call(json: Dict[str, Any]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs a JSON RPC call.
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.register_function(lambda **y: int('1', **y), 'int1')
>>> JsonRpcServer.execute_call([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({'id': 1})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['1']})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'int1', 'params': {'base': 16}})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'method': 'int1', 'params': {'base': 16}})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['a']})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32603, 'message': 'Internal error'}}
>>>
- execute_calls(json: Union[Dict[str, ~Json], List[Dict[str, ~Json]]]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs JSON RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.execute_calls([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(None)
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(Request('2.0', 'dir').asdict())
>>> JsonRpcServer.execute_calls([Request('2.0', 'builtins.int', 1, [1]).asdict()])
[{'jsonrpc': '2.0', 'id': 1, 'result': 1}]
>>>
- register_function(function: Callable, name: str = None) -> None from builtins.type
- This function adds a new function in the JSON RPC calls.
Data descriptors inherited from JsonRpcServer:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes inherited from JsonRpcServer:
- functions = {}
|
class BinaryJsonRpcRequestFormat(BinaryJsonFormat) |
| |
BinaryJsonRpcRequestFormat(method: Union[int, str], args: Optional[List[~Json]], kwargs: Optional[Dict[str, ~Json]], interface_version: int = None, encoding: str = 'latin1', id_: int = None, auth: ~Json = None, get_content: Callable = None)
This class implements the binary JSON RPC request format for a call.
|
| |
- Method resolution order:
- BinaryJsonRpcRequestFormat
- BinaryJsonFormat
- builtins.object
Methods defined here:
- __init__(self, method: Union[int, str], args: Optional[List[~Json]], kwargs: Optional[Dict[str, ~Json]], interface_version: int = None, encoding: str = 'latin1', id_: int = None, auth: ~Json = None, get_content: Callable = None)
- Initialize self. See help(type(self)) for accurate signature.
- asdict(self) -> Dict[str, ~Json]
- This method returns the JSON RPC from this instance.
>>> BinaryJsonRpcRequestFormat('test', [1, 2, 3], {'a': 'a'}, id_=1).asdict()
{'jsonrpc': 'py2.0', 'method': 'test', 'id': 1, 'args': [1, 2, 3], 'kwargs': {'a': 'a'}}
>>>
- load_params(self) -> None
- This method parses arguments data (bytes) to JSON.
- loads_pyparams(self) -> None
- This method parses arguments and keywords arguments data (bytes)
to a list and a dict.
- to_bytes(self) -> bytes
- This method returns bytes for this instance.
>>> BinaryJsonRpcRequestFormat('abc', [1, 2, 3], None, 5).to_bytes()
bytearray(b'\xc0\x03\x05Cabc\x83\x03\x01\x03\x02\x03\x03')
>>> BinaryJsonRpcRequestFormat('abc', None, {'a': 1, 'b': 2, 'c': 3}).to_bytes()
bytearray(b'\x80Cabc#Aa\x03\x01Ab\x03\x02Ac\x03\x03')
>>> BinaryJsonRpcRequestFormat(255, [1, 2, 3], {'a': 1, 'b': 2, 'c': 3}).to_bytes()
bytearray(b'\xa0\x03\xff\x83\x03\x01\x03\x02\x03\x03#Aa\x03\x01Ab\x03\x02Ac\x03\x03')
>>> BinaryJsonRpcRequestFormat(255, [3.14, 3.1425657895545824, [1], {1: 2}], None).to_bytes()
bytearray(b'\x80\x03\xff\x84\x0c\x1f\x85\xebQ\xb8\x1e\t@\x0c\xa3R]\x88\xf9#\t@\x81\x03\x01!\x03\x01\x03\x02')
>>> BinaryJsonRpcRequestFormat(255, None, {1: [1, [0], {2: 3, True: "abc"}, None, False]}).to_bytes()
bytearray(b'\x80\x03\xff!\x03\x01\x85\x03\x01\x81\x03\x00"\x03\x02\x03\x03\x01Cabc\x00\x02')
>>> BinaryJsonRpcRequestFormat(2 ** 0, [], None).to_bytes()
bytearray(b'\x80\x03\x01\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 8, [], None).to_bytes()
bytearray(b'\x80\x04\x01\x00\x80')
>>> BinaryJsonRpcRequestFormat(-1, [], None).to_bytes()
bytearray(b'\x80\x05\x00\x00\x00\x01\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 24, [], None).to_bytes()
bytearray(b'\x80\x06\x01\x00\x00\x00\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 32, [], None).to_bytes()
bytearray(b'\x80\x07\x01\x00\x00\x00\x00\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 40, [], None).to_bytes()
bytearray(b'\x80\x08\x01\x00\x00\x00\x00\x00\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 56 * -1, [], None).to_bytes()
bytearray(b'\x80\t\x01\x00\x00\x00\x00\x00\x00\x00\x80')
>>> BinaryJsonRpcRequestFormat(2 ** 56, [], None).to_bytes()
bytearray(b'\x80\n\x01\x00\x00\x00\x00\x00\x00\x00\x80')
>>> BinaryJsonRpcRequestFormat(0, [datetime(2025, 6, 22)], None).to_bytes()
bytearray(b'\x80\x03\x00\x81\x1dhW+`')
>>> BinaryJsonRpcRequestFormat(0, [timedelta(days=5, hours=1, minutes=30, microseconds=500)], None).to_bytes()
bytearray(b'\x80\x03\x00\x81\x1f\x00\x00\x00e\xd7\x12W\xf4')
>>>
Class methods defined here:
- from_bytes(data: bytes, get_content: Callable = None) -> ~BinaryJsonRpcRequestFormat from builtins.type
- This method builds an instance for this class from data.
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\xc0\x03\x05\r\x03abc\x19\x03\x03\x01\x03\x02\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=True, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.interface_version
5
>>> rpc.method
'abc'
>>> rpc.params
[1, 2, 3]
>>> rpc.args
[1, 2, 3]
>>> rpc.kwargs
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\r\x03abc\x15\x03\r\x01a\x03\x01\r\x01b\x03\x02\r\x01c\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
'abc'
>>> rpc.params
{'a': 1, 'b': 2, 'c': 3}
>>> rpc.args
>>> rpc.kwargs
{'a': 1, 'b': 2, 'c': 3}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\xa0\x03\xff\x19\x03\x03\x01\x03\x02\x03\x03\x15\x03\r\x01a\x03\x01\r\x01b\x03\x02\r\x01c\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=True, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
>>> rpc.args
[1, 2, 3]
>>> rpc.kwargs
{'a': 1, 'b': 2, 'c': 3}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\xff\x19\x04\x0c\x1f\x85\xebQ\xb8\x1e\t@\x0c\xa3R]\x88\xf9#\t@\x19\x01\x03\x01\x15\x01\x03\x01\x03\x02')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
[3.14, 3.1425657895545824, [1], {1: 2}]
>>> rpc.args
[3.14, 3.1425657895545824, [1], {1: 2}]
>>> rpc.kwargs
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\xff\x15\x01\x03\x01\x19\x05\x03\x01\x19\x01\x03\x00\x15\x02\x03\x02\x03\x03\x01\r\x03abc\x00\x02')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
{1: [1, [0], {2: 3, True: 'abc'}, None, False]}
>>> rpc.args
>>> rpc.kwargs
{1: [1, [0], {2: 3, True: 'abc'}, None, False]}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\xc0\x03\x05Cabc\x83\x03\x01\x03\x02\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=True, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.interface_version
5
>>> rpc.method
'abc'
>>> rpc.params
[1, 2, 3]
>>> rpc.args
[1, 2, 3]
>>> rpc.kwargs
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80Cabc#Aa\x03\x01Ab\x03\x02Ac\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
'abc'
>>> rpc.params
{'a': 1, 'b': 2, 'c': 3}
>>> rpc.args
>>> rpc.kwargs
{'a': 1, 'b': 2, 'c': 3}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\xa0\x03\xff\x83\x03\x01\x03\x02\x03\x03#Aa\x03\x01Ab\x03\x02Ac\x03\x03')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=True, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
>>> rpc.args
[1, 2, 3]
>>> rpc.kwargs
{'a': 1, 'b': 2, 'c': 3}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\xff\x84\x0c\x1f\x85\xebQ\xb8\x1e\t@\x0c\xa3R]\x88\xf9#\t@\x81\x03\x01!\x03\x01\x03\x02')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
[3.14, 3.1425657895545824, [1], {1: 2}]
>>> rpc.args
[3.14, 3.1425657895545824, [1], {1: 2}]
>>> rpc.kwargs
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\xff!\x03\x01\x85\x03\x01\x81\x03\x00"\x03\x02\x03\x03\x01Cabc\x00\x02')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
255
>>> rpc.params
{1: [1, [0], {2: 3, True: 'abc'}, None, False]}
>>> rpc.args
>>> rpc.kwargs
{1: [1, [0], {2: 3, True: 'abc'}, None, False]}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\x00\x81\x1dhW+`')
>>> rpc.headers
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>> rpc.method
0
>>> rpc.params
[datetime.datetime(2025, 6, 22, 0, 0)]
>>> rpc.args
[datetime.datetime(2025, 6, 22, 0, 0)]
>>> rpc.kwargs
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\x00\x81\x1f\x00\x00\x00e\xd7\x12W\xf4')
>>> rpc.params
[datetime.timedelta(days=5, seconds=5400, microseconds=500)]
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\x00\x81\x1e\x01\xdb\xe2\xf7\xd5>p\x00')
>>> rpc.args
[datetime.datetime(2025, 6, 22, 0, 0)]
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\x01\x80')
>>> rpc.method
1
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x04\x01\x00\x80')
>>> rpc.method
256
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x05\x00\x00\x00\x01\x80')
>>> rpc.method
-1
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x06\x01\x00\x00\x00\x80')
>>> rpc.method
16777216
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x07\x01\x00\x00\x00\x00\x80')
>>> rpc.method
4294967296
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x08\x01\x00\x00\x00\x00\x00\x80')
>>> rpc.method
1099511627776
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\t\x01\x00\x00\x00\x00\x00\x00\x00\x80')
>>> rpc.method
-72057594037927936
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\n\x01\x00\x00\x00\x00\x00\x00\x00\x80')
>>> rpc.method
72057594037927936
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\r\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x0e\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x0f\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x10\0\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x11\0\0\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x12\0\0\0\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x13\0\0\0\0\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x14\0\0\0\0\0\0\0\x01a\x80')
>>> rpc.method
'a'
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x15\1\0\0')
>>> rpc.params
{None: None}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x16\0\1\0\0')
>>> rpc.params
{None: None}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x17\0\0\1\0\0')
>>> rpc.params
{None: None}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x18\0\0\0\1\0\0')
>>> rpc.params
{None: None}
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x19\1\0')
>>> rpc.params
[None]
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x1a\0\1\0')
>>> rpc.params
[None]
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x1b\0\0\1\0')
>>> rpc.params
[None]
>>> rpc = BinaryJsonRpcRequestFormat.from_bytes(b'\x80\x03\0\x1c\0\0\0\1\0')
>>> rpc.params
[None]
>>>
Methods inherited from BinaryJsonFormat:
- dict_block(self, size_length: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- dict_dump(self, value: Dict[str, ~Json], data: bytearray) -> None
- This method dumps a Json like dict into binary format.
- dumps_data(self, data: ~Json, data_dump: bytearray) -> None
- This method encodes Json like data in binary format.
- duration(self) -> datetime.timedelta
- This method returns timedelta from binary data.
- duration_dump(self, value: datetime.timedelta) -> bytes
- This function dumps a timedelta (duration) to data bytes.
- float_dump(self, data: float) -> bytes
- This function dumps the float value.
- int_block(self, size: int, signed: bool = True) -> int
- This method transfrom the next block value into an integer.
- int_to_block(self, value: int, identifier: bool = True) -> Tuple[bytes, int]
- This method returns a block for a integer value.
- list_block(self, size_length: int) -> List[~Json]
- This method parses a Json like list from binary data.
- list_dump(self, value: List[~Json], data: bytearray) -> None
- This method dumps a Json like list into binary format.
- read(self, number: int) -> bytes
- This method reads `number` of bytes.
- read_dict(self, size: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- read_float4(self) -> float
- This function parses next 4 bytes as float.
- read_float8(self) -> float
- This function parses next 8 bytes as float.
- read_list(self, size: int) -> List[~Json]
- This method parses a Json like list from binary data.
- read_next_element(self) -> ~Json
- This function reads the next Json like element from binary data.
- read_str(self, size: int) -> str
- This method returns strings value from the binary data.
- string_block(self, size_length: int) -> str
- This method transfrom the next block value into an integer.
- string_to_block(self, value: str) -> bytes
- This method returns a block for a string value.
- unix_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to unix data bytes.
- unix_timestamp(self) -> datetime.datetime
- This method returns datetime for a unix timestamp from binary data.
- windows_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to windows data bytes.
>>> BinaryJsonFormat.windows_datetime_to_dump(None, datetime(2025, 6, 22))
b'\x1e\x01\xdb\xe2\xf7\xd5>p\x00'
>>>
- windows_timestamp(self) -> datetime.datetime
- This method returns datetime for a windows timestamp from binary data.
- write_block_size(self, element: ~Json, base: int, bytemap: Dict[int, bytes]) -> Tuple[bytes, bool]
- This method returns a block size and a boolean
to represent the usage of base.
Data descriptors inherited from BinaryJsonFormat:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
|
class BinaryJsonRpcResponseFormat(BinaryJsonFormat) |
| |
BinaryJsonRpcResponseFormat(id_: int, error: bool, content: ~Json, encoding: str = 'latin1', get_content: Callable = None)
This class implements the binary JSON RPC response format for a call.
|
| |
- Method resolution order:
- BinaryJsonRpcResponseFormat
- BinaryJsonFormat
- builtins.object
Methods defined here:
- __init__(self, id_: int, error: bool, content: ~Json, encoding: str = 'latin1', get_content: Callable = None)
- Initialize self. See help(type(self)) for accurate signature.
- to_bytes(self) -> bytes
- This method returns bytes for this instance.
>>> BinaryJsonRpcResponseFormat(5, True, {"code": 32600, "message": "Invalid Request"}).to_bytes()
bytearray(b'\x80\x03\x05"Dcode\x04\x7fXGmessageOInvalid Request')
>>> BinaryJsonRpcResponseFormat(2, False, 5).to_bytes()
bytearray(b'\x00\x03\x02\x03\x05')
>>> BinaryJsonRpcResponseFormat(1, False, {"code": 200, "stdout": "output", "stderr": "error"}).to_bytes()
bytearray(b'\x00\x03\x01#Dcode\x03\xc8FstdoutFoutputFstderrEerror')
>>> BinaryJsonRpcResponseFormat(1, False, [{"code": 200, "stdout": "output", "stderr": "error"}, {"code": 200, "stdout": "output", "stderr": "error"}]).to_bytes()
bytearray(b'\x00\x03\x01\x82#Dcode\x03\xc8FstdoutFoutputFstderrEerror#Dcode\x03\xc8FstdoutFoutputFstderrEerror')
>>>
Class methods defined here:
- from_bytes(data: bytes, get_content: Callable = None) -> ~BinaryJsonRpcResponseFormat from builtins.type
- This method builds an instance for this class from data.
>>> r = BinaryJsonRpcResponseFormat.from_bytes(b'\x80\x03\x05"Dcode\x04\x7fXGmessageOInvalid Request')
>>> r.call_id
5
>>> r.headers
ResponseHeaders(error=True)
>>> r.status
True
>>> r.content
{'code': 32600, 'message': 'Invalid Request'}
>>>
Methods inherited from BinaryJsonFormat:
- dict_block(self, size_length: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- dict_dump(self, value: Dict[str, ~Json], data: bytearray) -> None
- This method dumps a Json like dict into binary format.
- dumps_data(self, data: ~Json, data_dump: bytearray) -> None
- This method encodes Json like data in binary format.
- duration(self) -> datetime.timedelta
- This method returns timedelta from binary data.
- duration_dump(self, value: datetime.timedelta) -> bytes
- This function dumps a timedelta (duration) to data bytes.
- float_dump(self, data: float) -> bytes
- This function dumps the float value.
- int_block(self, size: int, signed: bool = True) -> int
- This method transfrom the next block value into an integer.
- int_to_block(self, value: int, identifier: bool = True) -> Tuple[bytes, int]
- This method returns a block for a integer value.
- list_block(self, size_length: int) -> List[~Json]
- This method parses a Json like list from binary data.
- list_dump(self, value: List[~Json], data: bytearray) -> None
- This method dumps a Json like list into binary format.
- read(self, number: int) -> bytes
- This method reads `number` of bytes.
- read_dict(self, size: int) -> Dict[str, ~Json]
- This method parses a Json like dict from binary data.
- read_float4(self) -> float
- This function parses next 4 bytes as float.
- read_float8(self) -> float
- This function parses next 8 bytes as float.
- read_list(self, size: int) -> List[~Json]
- This method parses a Json like list from binary data.
- read_next_element(self) -> ~Json
- This function reads the next Json like element from binary data.
- read_str(self, size: int) -> str
- This method returns strings value from the binary data.
- string_block(self, size_length: int) -> str
- This method transfrom the next block value into an integer.
- string_to_block(self, value: str) -> bytes
- This method returns a block for a string value.
- unix_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to unix data bytes.
- unix_timestamp(self) -> datetime.datetime
- This method returns datetime for a unix timestamp from binary data.
- windows_datetime_to_dump(self, value: datetime.datetime) -> bytes
- This function dumps a datetime to windows data bytes.
>>> BinaryJsonFormat.windows_datetime_to_dump(None, datetime(2025, 6, 22))
b'\x1e\x01\xdb\xe2\xf7\xd5>p\x00'
>>>
- windows_timestamp(self) -> datetime.datetime
- This method returns datetime for a windows timestamp from binary data.
- write_block_size(self, element: ~Json, base: int, bytemap: Dict[int, bytes]) -> Tuple[bytes, bool]
- This method returns a block size and a boolean
to represent the usage of base.
Data descriptors inherited from BinaryJsonFormat:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
|
class CsvRpcServer(_CsvRpcServerInterface, JsonRpcServer) |
| |
This class implements a CSV RPC.
>>> CsvRpcServer.register_function(dir)
>>> CsvRpcServer.register_function(int)
>>> CsvRpcServer.handle_request_data(b'"jsonrpc","method"\n"2.0","dir"\n')
>>> CsvRpcServer.handle_request_data(b'"jsonrpc","method","int?id","#0"\n"2.0","builtins.int","1","1"\n')
'"str?jsonrpc","int?id","int?result"\r\n"2.0","1","1"\r\n'
>>> CsvRpcServer.handle_request_data(b'"jsonrpc')
'"str?jsonrpc","NoneType?id","int?error.code","str?error.message"\r\n"2.0","None","-32600","Invalid Request"\r\n'
>>>
|
| |
- Method resolution order:
- CsvRpcServer
- _CsvRpcServerInterface
- JsonRpcServer
- builtins.object
Data and other attributes defined here:
- __annotations__ = {}
- pyrpc = False
Class methods inherited from _CsvRpcServerInterface:
- handle_request_data(csv: bytes) -> Optional[str] from builtins.type
- This method parses CSV from data to perform RPC calls.
Data descriptors inherited from _CsvRpcServerInterface:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Class methods inherited from JsonRpcServer:
- call(method: Callable, json: Dict[str, Any]) -> Any from builtins.type
- This function calls the python function.
- check_request(json: Dict[str, Any], id: Union[str, int, NoneType]) -> Optional[Dict[str, Union[str, int, JsonRpcExtended.Error]]] from builtins.type
- This function checks the JSON RPC received requests.
>>> JsonRpcServer.register_function(dir, 'test')
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.check_request({}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0'}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'test', 'params': None}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'builtins.dir'}, 1)
>>>
- execute_call(json: Dict[str, Any]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs a JSON RPC call.
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.register_function(lambda **y: int('1', **y), 'int1')
>>> JsonRpcServer.execute_call([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({'id': 1})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['1']})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'int1', 'params': {'base': 16}})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'method': 'int1', 'params': {'base': 16}})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['a']})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32603, 'message': 'Internal error'}}
>>>
- execute_calls(json: Union[Dict[str, ~Json], List[Dict[str, ~Json]]]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs JSON RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.execute_calls([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(None)
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(Request('2.0', 'dir').asdict())
>>> JsonRpcServer.execute_calls([Request('2.0', 'builtins.int', 1, [1]).asdict()])
[{'jsonrpc': '2.0', 'id': 1, 'result': 1}]
>>>
- register_function(function: Callable, name: str = None) -> None from builtins.type
- This function adds a new function in the JSON RPC calls.
Data and other attributes inherited from JsonRpcServer:
- functions = {}
|
class Error(builtins.object) |
| |
Error(code: int, message: str, data: str = None) -> None
Error(code: int, message: str, data: str = None)
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, code: int, message: str, data: str = None) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- asdict(self) -> Dict[str, Union[int, str]]
- This method returns this class as a dict for JsonRpc response.
>>> Error(0, "test", None).asdict()
{'code': 0, 'message': 'test'}
>>> Error(0, "test", "test2").asdict()
{'code': 0, 'message': 'test', 'data': 'test2'}
>>>
- checks(self) -> None
- This method checks data type.
>>> Error(0, "test", "test").checks()
>>> Error(0, "test", None).checks()
>>> Error("test", "test", "test").checks()
Traceback (most recent call last):
...
TypeError: Invalid code type
>>> Error(0, 0, "test").checks()
Traceback (most recent call last):
...
TypeError: Invalid message type
>>> Error(0, "", 0).checks()
Traceback (most recent call last):
...
TypeError: Invalid data type
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'code': <class 'int'>, 'data': <class 'str'>, 'message': <class 'str'>}
- __dataclass_fields__ = {'code': Field(name='code',type=<class 'int'>,default=<da...appingproxy({}),kw_only=False,_field_type=_FIELD), 'data': Field(name='data',type=<class 'str'>,default=Non...appingproxy({}),kw_only=False,_field_type=_FIELD), 'message': Field(name='message',type=<class 'str'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('code', 'message', 'data')
- data = None
|
class FORMATS(enum.IntEnum) |
| |
FORMATS(value, names=None, *, module=None, qualname=None, type=None, start=1)
An enumeration.
|
| |
- Method resolution order:
- FORMATS
- enum.IntEnum
- builtins.int
- enum.Enum
- builtins.object
Data and other attributes defined here:
- BINARY = <FORMATS.BINARY: 3>
- CSV = <FORMATS.CSV: 1>
- JSON = <FORMATS.JSON: 0>
- XML = <FORMATS.XML: 2>
Data descriptors inherited from enum.Enum:
- name
- The name of the Enum member.
- value
- The value of the Enum member.
Readonly properties inherited from enum.EnumMeta:
- __members__
- Returns a mapping of member name->value.
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.
|
class JsonRpcServer(builtins.object) |
| |
This class implements JSON RPC RFC.
Multiples methods come from: https://github.com/mauricelambert/WebScripts/blob/main/WebScripts/modules/JsonRpc.py
|
| |
Class methods defined here:
- call(method: Callable, json: Dict[str, Any]) -> Any from builtins.type
- This function calls the python function.
- check_request(json: Dict[str, Any], id: Union[str, int, NoneType]) -> Optional[Dict[str, Union[str, int, JsonRpcExtended.Error]]] from builtins.type
- This function checks the JSON RPC received requests.
>>> JsonRpcServer.register_function(dir, 'test')
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.check_request({}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0'}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'test', 'params': None}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> JsonRpcServer.check_request({'jsonrpc': '2.0', 'method': 'builtins.dir'}, 1)
>>>
- execute_call(json: Dict[str, Any]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs a JSON RPC call.
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.register_function(lambda **y: int('1', **y), 'int1')
>>> JsonRpcServer.execute_call([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({'id': 1})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['1']})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'int1', 'params': {'base': 16}})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'method': 'int1', 'params': {'base': 16}})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['a']})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32603, 'message': 'Internal error'}}
>>>
- execute_calls(json: Union[Dict[str, ~Json], List[Dict[str, ~Json]]]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs JSON RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.execute_calls([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(None)
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(Request('2.0', 'dir').asdict())
>>> JsonRpcServer.execute_calls([Request('2.0', 'builtins.int', 1, [1]).asdict()])
[{'jsonrpc': '2.0', 'id': 1, 'result': 1}]
>>>
- handle_request_data(json: bytes, permissive: bool = False) -> Optional[str] from builtins.type
- This function parses JSON from data to perform RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc": "2.0", "method": "dir"}')
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc" "2.0" "method" "builtins.int", "id": 1 "params" ["1"],,,}', permissive=True)
'{"jsonrpc": "2.0", "id": 1, "result": 1}'
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc" ')
'{"jsonrpc": "2.0", "id": null, "error": {"code": -32600, "message": "Invalid Request"}}'
>>>
- register_function(function: Callable, name: str = None) -> None from builtins.type
- This function adds a new function in the JSON RPC calls.
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'functions': typing.Dict[str, typing.Callable]}
- functions = {}
|
class PyCsvRpcServer(_CsvRpcServerInterface, PyJsonRpcServer) |
| |
This class implements a modified CSV RPC.
>>> PyCsvRpcServer.register_function(dir)
>>> PyCsvRpcServer.register_function(int)
>>> PyCsvRpcServer.handle_request_data(b'"jsonrpc","method"\n"py2.0","dir"\n')
>>> PyCsvRpcServer.handle_request_data(b'"jsonrpc","method","int?id","#0"\n"py2.0","builtins.int","1","1"\n')
'"str?jsonrpc","int?id","int?result"\r\n"2.0","1","1"\r\n'
>>> PyCsvRpcServer.handle_request_data(b'"jsonrpc')
'"str?jsonrpc","NoneType?id","int?error.code","str?error.message"\r\n"2.0","None","-32600","Invalid Request"\r\n'
>>>
|
| |
- Method resolution order:
- PyCsvRpcServer
- _CsvRpcServerInterface
- PyJsonRpcServer
- JsonRpcServer
- builtins.object
Data and other attributes defined here:
- __annotations__ = {}
- pyrpc = True
Class methods inherited from _CsvRpcServerInterface:
- handle_request_data(csv: bytes) -> Optional[str] from builtins.type
- This method parses CSV from data to perform RPC calls.
Data descriptors inherited from _CsvRpcServerInterface:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Class methods inherited from PyJsonRpcServer:
- call(method: Callable, json: Dict[str, Any]) -> Any from builtins.type
- This function calls the python function.
- check_request(json: Dict[str, Any], id: Union[str, int, NoneType]) -> Optional[Dict[str, Union[str, int, JsonRpcExtended.Error]]] from builtins.type
- This function checks the JSON RPC received requests.
>>> PyJsonRpcServer.register_function(dir, 'test')
>>> PyJsonRpcServer.register_function(dir)
>>> PyJsonRpcServer.check_request({}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0'}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'test', 'args': {}}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'test', 'kwargs': []}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'builtins.dir'}, 1)
>>>
Class methods inherited from JsonRpcServer:
- execute_call(json: Dict[str, Any]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs a JSON RPC call.
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.register_function(lambda **y: int('1', **y), 'int1')
>>> JsonRpcServer.execute_call([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({'id': 1})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['1']})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'int1', 'params': {'base': 16}})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'method': 'int1', 'params': {'base': 16}})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['a']})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32603, 'message': 'Internal error'}}
>>>
- execute_calls(json: Union[Dict[str, ~Json], List[Dict[str, ~Json]]]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs JSON RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.execute_calls([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(None)
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(Request('2.0', 'dir').asdict())
>>> JsonRpcServer.execute_calls([Request('2.0', 'builtins.int', 1, [1]).asdict()])
[{'jsonrpc': '2.0', 'id': 1, 'result': 1}]
>>>
- register_function(function: Callable, name: str = None) -> None from builtins.type
- This function adds a new function in the JSON RPC calls.
Data and other attributes inherited from JsonRpcServer:
- functions = {}
|
class PyJsonRpcServer(JsonRpcServer) |
| |
This class implements a modified JSON RPC for python.
|
| |
- Method resolution order:
- PyJsonRpcServer
- JsonRpcServer
- builtins.object
Class methods defined here:
- call(method: Callable, json: Dict[str, Any]) -> Any from builtins.type
- This function calls the python function.
- check_request(json: Dict[str, Any], id: Union[str, int, NoneType]) -> Optional[Dict[str, Union[str, int, JsonRpcExtended.Error]]] from builtins.type
- This function checks the JSON RPC received requests.
>>> PyJsonRpcServer.register_function(dir, 'test')
>>> PyJsonRpcServer.register_function(dir)
>>> PyJsonRpcServer.check_request({}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0'}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'test', 'args': {}}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'test', 'kwargs': []}, 1)
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32602, 'message': 'Invalid params'}}
>>> PyJsonRpcServer.check_request({'jsonrpc': 'py2.0', 'method': 'builtins.dir'}, 1)
>>>
Data and other attributes defined here:
- __annotations__ = {}
Class methods inherited from JsonRpcServer:
- execute_call(json: Dict[str, Any]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs a JSON RPC call.
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.register_function(lambda **y: int('1', **y), 'int1')
>>> JsonRpcServer.execute_call([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({'id': 1})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_call({})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['1']})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'int1', 'params': {'base': 16}})
{'jsonrpc': '2.0', 'id': 1, 'result': 1}
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'method': 'int1', 'params': {'base': 16}})
>>> JsonRpcServer.execute_call({'jsonrpc': '2.0', 'id': 1, 'method': 'builtins.int', 'params': ['a']})
{'jsonrpc': '2.0', 'id': 1, 'error': {'code': -32603, 'message': 'Internal error'}}
>>>
- execute_calls(json: Union[Dict[str, ~Json], List[Dict[str, ~Json]]]) -> Optional[Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]] from builtins.type
- This function performs JSON RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.execute_calls([])
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(None)
{'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid Request'}}
>>> JsonRpcServer.execute_calls(Request('2.0', 'dir').asdict())
>>> JsonRpcServer.execute_calls([Request('2.0', 'builtins.int', 1, [1]).asdict()])
[{'jsonrpc': '2.0', 'id': 1, 'result': 1}]
>>>
- handle_request_data(json: bytes, permissive: bool = False) -> Optional[str] from builtins.type
- This function parses JSON from data to perform RPC calls.
>>> JsonRpcServer.register_function(dir)
>>> JsonRpcServer.register_function(int)
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc": "2.0", "method": "dir"}')
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc" "2.0" "method" "builtins.int", "id": 1 "params" ["1"],,,}', permissive=True)
'{"jsonrpc": "2.0", "id": 1, "result": 1}'
>>> JsonRpcServer.handle_request_data(b'{"jsonrpc" ')
'{"jsonrpc": "2.0", "id": null, "error": {"code": -32600, "message": "Invalid Request"}}'
>>>
- register_function(function: Callable, name: str = None) -> None from builtins.type
- This function adds a new function in the JSON RPC calls.
Data descriptors inherited from JsonRpcServer:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes inherited from JsonRpcServer:
- functions = {}
|
class PyRequest(builtins.object) |
| |
PyRequest(jsonrpc: str, method: str, id: Union[str, int, NoneType] = None, args: List[~Json] = None, kwargs: Dict[str, ~Json] = None) -> None
This dataclass implements the modified JSON RPC request for python.
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, jsonrpc: str, method: str, id: Union[str, int, NoneType] = None, args: List[~Json] = None, kwargs: Dict[str, ~Json] = None) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- asdict(self) -> Dict[str, Union[int, str]]
- This method returns this class as a dict for JsonRpc request.
>>> PyRequest("", "", None, None).asdict()
{'jsonrpc': '', 'method': ''}
>>> PyRequest("py2.0", "test", 1, [1, 2, 3], {'a': 'a'}).asdict()
{'jsonrpc': 'py2.0', 'method': 'test', 'id': 1, 'args': [1, 2, 3], 'kwargs': {'a': 'a'}}
>>>
- checks(self) -> None
- This method checks data type.
>>> PyRequest("2.0", "test", 1, [], {}).checks()
>>> PyRequest(0, "test", "test", []).checks()
Traceback (most recent call last):
...
TypeError: Invalid jsonrpc type
>>> PyRequest("", 0, "test", []).checks()
Traceback (most recent call last):
...
TypeError: Invalid method type
>>> PyRequest("", "", [], []).checks()
Traceback (most recent call last):
...
TypeError: Invalid id type
>>> PyRequest("", "", "", 0, {}).checks()
Traceback (most recent call last):
...
TypeError: Invalid args type
>>> PyRequest("", "", "", [], 0).checks()
Traceback (most recent call last):
...
TypeError: Invalid kwargs type
>>> PyRequest("", "", "test", [], {}).checks()
>>> PyRequest("", "", None, None, None).checks()
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'args': typing.List[~Json], 'id': typing.Union[str, int, NoneType], 'jsonrpc': <class 'str'>, 'kwargs': typing.Dict[str, ~Json], 'method': <class 'str'>}
- __dataclass_fields__ = {'args': Field(name='args',type=typing.List[~Json],defaul...appingproxy({}),kw_only=False,_field_type=_FIELD), 'id': Field(name='id',type=typing.Union[str, int, None...appingproxy({}),kw_only=False,_field_type=_FIELD), 'jsonrpc': Field(name='jsonrpc',type=<class 'str'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD), 'kwargs': Field(name='kwargs',type=typing.Dict[str, ~Json]...appingproxy({}),kw_only=False,_field_type=_FIELD), 'method': Field(name='method',type=<class 'str'>,default=<...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('jsonrpc', 'method', 'id', 'args', 'kwargs')
- args = None
- id = None
- kwargs = None
|
class Request(builtins.object) |
| |
Request(jsonrpc: str, method: str, id: Union[str, int, NoneType] = None, params: Union[List[~Json], Dict[str, ~Json]] = None) -> None
This dataclass implements the JSON RPC request.
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, jsonrpc: str, method: str, id: Union[str, int, NoneType] = None, params: Union[List[~Json], Dict[str, ~Json]] = None) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- asdict(self) -> Dict[str, Union[int, str]]
- This method returns this class as a dict for JsonRpc request.
>>> Request("", "", None, None).asdict()
{'jsonrpc': '', 'method': ''}
>>> Request("2.0", "test", 1, [1, 2, 3]).asdict()
{'jsonrpc': '2.0', 'method': 'test', 'id': 1, 'params': [1, 2, 3]}
>>>
- checks(self) -> None
- This method checks data type.
>>> Request("2.0", "test", "test", []).checks()
>>> Request(0, "test", "test", []).checks()
Traceback (most recent call last):
...
TypeError: Invalid jsonrpc type
>>> Request("", 0, "test", []).checks()
Traceback (most recent call last):
...
TypeError: Invalid method type
>>> Request("", "", [], []).checks()
Traceback (most recent call last):
...
TypeError: Invalid id type
>>> Request("", "", "", 0).checks()
Traceback (most recent call last):
...
TypeError: Invalid params type
>>> Request("", "", "", None).checks()
>>> Request("", "", "", {}).checks()
>>> Request("", "", 0, {}).checks()
>>> Request("", "", None, {}).checks()
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'id': typing.Union[str, int, NoneType], 'jsonrpc': <class 'str'>, 'method': <class 'str'>, 'params': typing.Union[typing.List[~Json], typing.Dict[str, ~Json]]}
- __dataclass_fields__ = {'id': Field(name='id',type=typing.Union[str, int, None...appingproxy({}),kw_only=False,_field_type=_FIELD), 'jsonrpc': Field(name='jsonrpc',type=<class 'str'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD), 'method': Field(name='method',type=<class 'str'>,default=<...appingproxy({}),kw_only=False,_field_type=_FIELD), 'params': Field(name='params',type=typing.Union[typing.Lis...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('jsonrpc', 'method', 'id', 'params')
- id = None
- params = None
|
class RequestHeaders(builtins.object) |
| |
RequestHeaders(binary: bool, has_version: bool, py_json_rpc: bool, has_id: bool, has_auth: bool) -> None
RequestHeaders(binary: bool, has_version: bool, py_json_rpc: bool, has_id: bool, has_auth: bool)
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, binary: bool, has_version: bool, py_json_rpc: bool, has_id: bool, has_auth: bool) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- to_byte(self) -> int
- This method returns the byte for the headers.
>>> RequestHeaders(True, True, True, True, True).to_byte()
248
>>> bin(RequestHeaders(True, False, False, False, False).to_byte())
'0b10000000'
>>>
Class methods defined here:
- from_bytes(data: bytes) -> ~RequestHeaders from builtins.type
- This method builds an instance for this class from data.
>>> RequestHeaders.from_bytes(bytes([248]))
RequestHeaders(binary=True, has_version=True, py_json_rpc=True, has_id=True, has_auth=True)
>>> RequestHeaders.from_bytes(bytes([0b10000000]))
RequestHeaders(binary=True, has_version=False, py_json_rpc=False, has_id=False, has_auth=False)
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'binary': <class 'bool'>, 'has_auth': <class 'bool'>, 'has_id': <class 'bool'>, 'has_version': <class 'bool'>, 'py_json_rpc': <class 'bool'>}
- __dataclass_fields__ = {'binary': Field(name='binary',type=<class 'bool'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD), 'has_auth': Field(name='has_auth',type=<class 'bool'>,defaul...appingproxy({}),kw_only=False,_field_type=_FIELD), 'has_id': Field(name='has_id',type=<class 'bool'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD), 'has_version': Field(name='has_version',type=<class 'bool'>,def...appingproxy({}),kw_only=False,_field_type=_FIELD), 'py_json_rpc': Field(name='py_json_rpc',type=<class 'bool'>,def...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('binary', 'has_version', 'py_json_rpc', 'has_id', 'has_auth')
|
class Response(builtins.object) |
| |
Response(jsonrpc: str, id: Union[str, int, NoneType], result: ~Json = None, error: JsonRpcExtended.Error = None) -> None
This dataclass implements the JSON RPC request.
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, jsonrpc: str, id: Union[str, int, NoneType], result: ~Json = None, error: JsonRpcExtended.Error = None) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- asdict(self, id: Union[str, int] = None) -> Dict[str, Union[int, str, ~Json, JsonRpcExtended.Error]]
- This method returns this class as a dict for JsonRpc response.
>>> Response("", "", None, None).asdict()
{'jsonrpc': '', 'id': ''}
>>> Response("2.0", "test", [1, 2, 3]).asdict()
{'jsonrpc': '2.0', 'id': 'test', 'result': [1, 2, 3]}
>>> Response("2.0", "test", None, Error(0, 'test')).asdict()
{'jsonrpc': '2.0', 'id': 'test', 'error': {'code': 0, 'message': 'test'}}
>>>
- checks(self) -> None
- This method checks data type.
>>> Response("2.0", None, None, None).checks()
>>> Response("2.0", 1, {}, Error(0, "")).checks()
Traceback (most recent call last):
...
ValueError: Error and result should not be set in the same response.
>>> Response("2.0", 1, None, Error(0, "")).checks()
>>> Response("2.0", 1, {}, None).checks()
>>> Response(0, 1, {}, None).checks()
Traceback (most recent call last):
...
TypeError: Invalid jsonrpc type
>>> Response("", [], {}, None).checks()
Traceback (most recent call last):
...
TypeError: Invalid id type
>>> Response("", "", Error(0, ""), None).checks()
Traceback (most recent call last):
...
TypeError: Invalid result type
>>> Response("", "", {}, {}).checks()
Traceback (most recent call last):
...
TypeError: Invalid error type
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'error': <class 'JsonRpcExtended.Error'>, 'id': typing.Union[str, int, NoneType], 'jsonrpc': <class 'str'>, 'result': ~Json}
- __dataclass_fields__ = {'error': Field(name='error',type=<class 'JsonRpcExtended....appingproxy({}),kw_only=False,_field_type=_FIELD), 'id': Field(name='id',type=typing.Union[str, int, None...appingproxy({}),kw_only=False,_field_type=_FIELD), 'jsonrpc': Field(name='jsonrpc',type=<class 'str'>,default=...appingproxy({}),kw_only=False,_field_type=_FIELD), 'result': Field(name='result',type=~Json,default=None,defa...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('jsonrpc', 'id', 'result', 'error')
- error = None
- result = None
|
class ResponseHeaders(builtins.object) |
| |
ResponseHeaders(error: bool) -> None
ResponseHeaders(error: bool)
|
| |
Methods defined here:
- __eq__(self, other)
- Return self==value.
- __init__(self, error: bool) -> None
- Initialize self. See help(type(self)) for accurate signature.
- __repr__(self)
- Return repr(self).
- to_byte(self) -> int
- This method returns the byte for the headers.
>>> bin(ResponseHeaders(True).to_byte())
'0b10000000'
>>> ResponseHeaders(False).to_byte()
0
>>>
Class methods defined here:
- from_bytes(data: bytes) -> ~ResponseHeaders from builtins.type
- This method builds an instance for this class from data.
>>> ResponseHeaders.from_bytes(bytes([0b10000000]))
ResponseHeaders(error=True)
>>> ResponseHeaders.from_bytes(bytes([0b00000000]))
ResponseHeaders(error=False)
>>>
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
Data and other attributes defined here:
- __annotations__ = {'error': <class 'bool'>}
- __dataclass_fields__ = {'error': Field(name='error',type=<class 'bool'>,default=<...appingproxy({}),kw_only=False,_field_type=_FIELD)}
- __dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
- __hash__ = None
- __match_args__ = ('error',)
| |