Rockets python client
Project description
Rockets Python Client
A small client for Rockets using JSON-RPC as communication contract over a WebSocket.
Table of Contents
Installation
You can install this package from PyPI:
pip install rockets
Usage
Client
vs. AsyncClient
Rockets provides two types of clients to support asychronous and synchronous usage.
The AsyncClient
exposes all of its functionality as async
functions, hence an asyncio
event loop is needed to complete pending
execution via await
or run_until_complete()
.
For simplicity, a synchronous Client
is provided which automagically executes in a synchronous,
blocking fashion.
Connection
Create a client and connect:
from rockets import Client # client does not connect during __init__; # either explicit or automatically on any notify/request/send client = Client('myhost:8080') client.connect() print(client.connected())
Close the connection with the socket cleanly:
from rockets import Client client = Client('myhost:8080') client.connect() client.disconnect() print(client.connected())
Server messages
Listen to server notifications:
from rockets import Client client = Client('myhost:8080') client.notifications.subscribe(lambda msg: print("Got message:", msg.data))
NOTE: The notification object is of type Notification
.
Listen to any server message:
from rockets import Client client = Client('myhost:8080') client.ws_observable.subscribe(lambda msg: print("Got message:", msg))
Notifications
Send notifications to the server:
from rockets import Client client = Client('myhost:8080') client.notify('mymethod', {'ping': True})
Requests
Make a synchronous, blocking request:
from rockets import Client client = Client('myhost:8080') response = client.request('mymethod', {'ping': True}) print(response)
Handle a request error:
from rockets import Client, RequestError client = Client('myhost:8080') try: client.request('mymethod') except RequestError as err: print(err.code, err.message)
NOTE: Any error that may occur will be a RequestError
.
Asynchronous requests
Make an asynchronous request, using the AsyncClient
and asyncio
:
import asyncio from rockets import AsyncClient client = AsyncClient('myhost:8080') request_task = client.async_request('mymethod', {'ping': True}) asyncio.get_event_loop().run_until_complete(request_task) print(request_task.result())
Alternatively, you can use add_done_callback()
from the returned RequestTask
which is called
once the request has finished:
import asyncio from rockets import AsyncClient client = AsyncClient('myhost:8080') request_task = client.async_request('mymethod', {'ping': True}) request_task.add_done_callback(lambda task: print(task.result())) asyncio.get_event_loop().run_until_complete(request_task)
If the RequestTask
is not needed, i.e. no cancel()
or add_progress_callback()
is desired, use
the request()
coroutine:
import asyncio from rockets import AsyncClient client = AsyncClient('myhost:8080') coro = client.request('mymethod', {'ping': True}) result = asyncio.get_event_loop().run_until_complete(coro) print(result)
If you are already in an async
function or in a Jupyter notebook cell, you may use await
to
execute an asynchronous request:
# Inside a notebook cell here import asyncio from rockets import AsyncClient client = AsyncClient('myhost:8080') result = await client.request('mymethod', {'ping': True}) print(result)
Cancel a request:
from rockets import AsyncClient client = AsyncClient('myhost:8080') request_task = client.async_request('mymethod') request_task.cancel()
Get progress updates for a request:
from rockets import AsyncClient client = AsyncClient('myhost:8080') request_task = client.async_request('mymethod') request_task.add_progress_callback(lambda progress: print(progress))
NOTE: The progress object is of type RequestProgress
.
Batching
Make a batch request:
from rockets import Client, Request, Notification client = Client('myhost:8080') request = Request('myrequest') notification = Notification('mynotify') responses = client.batch([request, notification]) for response in responses: print(response)
Cancel a batch request:
from rockets import AsyncClient client = AsyncClient('myhost:8080') request = Request('myrequest') notification = Notification('mynotify') request_task = client.async_batch([request, notification]) request_task.cancel()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.