Django-channels-jsonrpc
For channels 2, see here
The Django-channels-jsonrpc is aimed to enable JSON-RPC functionnality on top of the excellent django channels project and especially their Websockets functionality. It is aimed to be:
- Fully integrated with Channels
- Fully implement JSON-RPC 1 and 2 protocol
- Support both WebSocket and HTTP transports
- Easy integration
Tech
The only Django-channels-jsonrpc dependency is the Django channels project
Installation
Download and extract the latest pre-built release.
Install the dependencies and devDependencies and start the server.
$ pip install django-channels-jsonrpc
Use
See complete example here, and in particular consumer.py
It is intended to be used as a Websocket consumer. See documentation except... simplier...
Import JsonRpcConsumer class and create the consumer
from channels_jsonrpc import JsonRpcConsumer
class MyJsonRpcConsumer(JsonRpcConsumer):
def connect(self, message, **kwargs):
"""
Perform things on WebSocket connection start
"""
self.message.reply_channel.send({"accept": True})
print("connect")
# Do stuff if needed
def disconnect(self, message, **kwargs):
"""
Perform things on WebSocket connection close
"""
print("disconnect")
# Do stuff if needed
JsonRpcConsumer derives from Channels WebSocketConsumer, you can read about all it's features here: https://channels.readthedocs.io/en/stable/generics.html#websockets
Then the last step is to create the RPC methos hooks. IT is done with the decorator:
@MyJsonRpcConsumer.rpc_method()
Like this:
@MyJsonRpcConsumer.rpc_method()
def ping():
return "pong"
MyJsonRpcConsumer.rpc_method() accept a string as a parameter to 'rename' the function
@MyJsonRpcConsumer.rpc_method("mymodule.rpc.ping")
def ping():
return "pong"
Will now be callable with "method":"mymodule.rpc.ping" in the rpc call:
{"id":1, "jsonrpc":"2.0","method":"mymodule.rpc.ping","params":{}}
RPC methods can obviously accept parameters. They also return "results" or "errors":
@MyJsonRpcConsumer.rpc_method("mymodule.rpc.ping")
def ping(fake_an_error):
if fake_an_error:
# Will return an error to the client
# --> {"id":1, "jsonrpc":"2.0","method":"mymodule.rpc.ping","params":{}}
# <-- {"id": 1, "jsonrpc": "2.0", "error": {"message": "fake_error", "code": -32000, "data": ["fake_error"]}}
raise Exception("fake_error")
else:
# Will return a result to the client
# --> {"id":1, "jsonrpc":"2.0","method":"mymodule.rpc.ping","params":{}}
# <-- {"id": 1, "jsonrpc": "2.0", "result": "pong"}
return "pong"
Sessions and other parameters from Message object
The original channel message - that can contain sessions (if activated with http_user) and other important info can be easily accessed by retrieving the **kwargs
and get a parameter named original_message
MyJsonRpcConsumerTest.rpc_method()
def json_rpc_method(param1, **kwargs):
original_message = kwargs["orginal_message"]
##do something with original_message
Example:
class MyJsonRpcConsumerTest(JsonRpcConsumer):
# Set to True to automatically port users from HTTP cookies
# (you don't need channel_session_user, this implies it)
# https://channels.readthedocs.io/en/stable/generics.html#websockets
http_user = True
....
@MyJsonRpcConsumerTest.rpc_method()
def ping(**kwargs):
original_message = kwargs["orginal_message"]
original_message.channel_session["test"] = True
return "pong"
Notifications
Inbound notifications
Those are the one sent from the client to the server.
They are dealt with the same way RPC methods are, except that instead of using rpc_method()
, you can use rpc_notification()
Thos rpc_notifications
can also retrieve the original_message
object
# Will be triggered when receiving this
# --> {"jsonrpc":"2.0","method":"notification.alt_name","params":["val_param1", "val_param2"]}
@MyJsonRpcWebsocketConsumerTest.rpc_notification("notification.alt_name")
def notification1(param1, param2, **kwargs):
original_message = kwargs["orginal_message"]
# Do something with notification
# ...
# Notification shouldn't return anything.
return
Outbound notifications
The server might want to send notifications to one or more of its clients. For that JsonRpcWebsocketConsumer
provides 2 static methods:
- JsonRpcWebsocketConsumer.notify_group(group_name, method, params)
Using channels'groups you can notify a whole group using this method
@MyJsonRpcWebsocketConsumerTest.rpc_method()
def send_to_group(group_name):
MyJsonRpcWebsocketConsumerTest.notify_group(group_name, "notification.notif", {"payload": 1234})
return True
Calling the RPC-method will send this notification to all the group group_name
- JsonRpcWebsocketConsumer.notify_channel(reply_channel, method, params)
This will notify only one channel/client.
@MyJsonRpcWebsocketConsumerTest.rpc_method()
def send_to_reply_channel(**kwargs):
original_message = kwarg["original_message"]
MyJsonRpcWebsocketConsumerTest.notify_channel(original_message.reply_channel,
"notification.ownnotif",
{"payload": 12})
return True
The reply_channel
can be found in theoriginal_message
object.
Transport-specific rpc-method/notifications
If you want to restrict rpc methods or notifications access to a specific transport method (http or websocket)
The two decorator rpc_method()
and rpc_notification()
accept parameters to restric their use. websocket
(default: True) and http
(default: True)
You can use them like this:
@MyJsonRpcWebsocketConsumerTest.rpc_notification("notification.alt_name", websocket=True, http=False)
def notification1(param1, param2, **kwargs):
original_message = kwargs["orginal_message"]
# This notification will only be used when using websocket transport
return
Custom JSON encoder class
from django.core.serializers.json import DjangoJSONEncoder
class DjangoJsonRpcConsumer(JsonRpcConsumer):
json_encoder_class = DjangoJSONEncoder
Testing
The JsonRpcConsumer class can be tested the same way Channels Consumers are tested. See here
You just need to remember to set your JsonRpcConsumer class to TEST_MODE in the test:
from channels.tests import ChannelTestCase, HttpClient
from .consumer import MyJsonRpcConsumer
MyJsonRpcConsumer.TEST_MODE = True
class TestsJsonConsumer(ChannelTestCase):
def assertResult(self, method, params, result, error=False):
client = HttpClient()
client.send_and_consume('websocket.receive', text=request(method, params))
key = "result" if not error else "error"
message = client.receive()
if message is None or key not in message:
raise KeyError("'%s' key not in message: %s" % (key, message))
self.assertEquals(message[key], result)
def assertError(self, method, params, result):
self.assertResult(method, params, result, True)
def test_assert_result(self):
self.assertResult("ping", {}, "pong")
License
MIT
Have fun with Websockets!
Free Software, Hell Yeah!