Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test(self):
assert Notification("get") == {"jsonrpc": "2.0", "method": "get"}
def test_positional(self):
assert Notification("sqrt", 1) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [1],
}
assert Notification("sqrt", [1, 2, 3]) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [[1, 2, 3]],
}
assert Notification("sqrt", {"name": "Foo"}) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [{"name": "Foo"}],
}
def test_str(self):
assert str(Notification("get")) == '{"jsonrpc": "2.0", "method": "get"}'
def test_both(self):
assert Notification("find", "Foo", age=42) == {
"jsonrpc": "2.0",
"method": "find",
"params": ["Foo", {"age": 42}],
}
def test_positional(self):
assert Notification("sqrt", 1) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [1],
}
assert Notification("sqrt", [1, 2, 3]) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [[1, 2, 3]],
}
assert Notification("sqrt", {"name": "Foo"}) == {
"jsonrpc": "2.0",
"method": "sqrt",
"params": [{"name": "Foo"}],
}
def test_keyword(self):
assert Notification("find", name="Foo") == {
"jsonrpc": "2.0",
"method": "find",
"params": {"name": "Foo"},
}
async def main():
async with websockets.connect("ws://localhost:5000") as ws:
requests = [Request("ping"), Notification("ping"), Request("ping")]
response = await WebSocketsClient(ws).send(requests)
for data in response.data:
if data.ok:
print("{}: {}".format(data.id, data.result))
else:
logging.error("%d: %s", data.id, data.message)
import logging
from jsonrpcclient.clients.http_client import HTTPClient
from jsonrpcclient.requests import Request, Notification
client = HTTPClient("http://localhost:5000")
response = client.send([Request("ping"), Notification("ping"), Request("ping")])
for data in response.data:
if data.ok:
print("{}: {}".format(data.id, data.result))
else:
logging.error("%d: %s", data.id, data.message)
# jsonrpcclient to keep compatibility.
# TODO: consider to raise a warning.
params_list = list(args)
params_list.append(kwargs)
self.update(params=params_list)
elif args:
self.update(params=list(args))
elif kwargs:
self.update(params=kwargs)
def __str__(self) -> str:
"""Wrapper around request, returning a string instead of a dict"""
return json.dumps(sort_request(self))
class Request(Notification):
"""
Create a JSON-RPC request object
http://www.jsonrpc.org/specification#request_object.
>>> Request("cat", name="Yoko")
{'jsonrpc': '2.0', 'method': 'cat', 'params': {'name': 'Yoko'}, 'id': 1}
Args:
method: The `method` name.
args: Positional arguments added to `params`.
kwargs: Keyword arguments added to `params`. Use request_id=x to force the
`id` to use.
Returns:
The JSON-RPC request in dictionary form.
"""