jsonrpcserver Examples¶
Showing how to take JSON-RPC requests in various frameworks and transport protocols.
aiohttp¶
$ pip install jsonrpcserver aiohttp
from aiohttp import web
from jsonrpcserver.aio import methods
@methods.add
async def ping():
return 'pong'
async def handle(request):
request = await request.text()
response = await methods.dispatch(request)
if response.is_notification:
return web.Response()
else:
return web.json_response(response, status=response.http_status)
app = web.Application()
app.router.add_post('/', handle)
if __name__ == '__main__':
web.run_app(app, port=5000)
See blog post.
Django¶
Create a views.py
:
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from jsonrpcserver import methods
@methods.add
def ping():
return 'pong'
@csrf_exempt
def jsonrpc(request):
response = methods.dispatch(request.body.decode())
return JsonResponse(response, status=response.http_status)
See blog post.
Flask¶
$ pip install jsonrpcserver flask
from flask import Flask, request, Response
from jsonrpcserver import methods
app = Flask(__name__)
@methods.add
def ping():
return 'pong'
@app.route('/', methods=['POST'])
def index():
req = request.get_data().decode()
response = methods.dispatch(req)
return Response(str(response), response.http_status,
mimetype='application/json')
if __name__ == '__main__':
app.run()
See blog post.
http.server¶
Python’s built-in http.server module.
$ pip install jsonrpcserver
from http.server import BaseHTTPRequestHandler, HTTPServer
from jsonrpcserver import methods
@methods.add
def ping():
return 'pong'
class TestHttpServer(BaseHTTPRequestHandler):
def do_POST(self):
# Process request
request = self.rfile.read(int(self.headers['Content-Length'])).decode()
response = methods.dispatch(request)
# Return response
self.send_response(response.http_status)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(str(response).encode())
if __name__ == '__main__':
HTTPServer(('localhost', 5000), TestHttpServer).serve_forever()
See blog post.
Plain jsonrpcserver¶
Using jsonrpcserver’s built-in serve_forever
method.
$ pip install jsonrpcserver
The quickest way to serve a method:
from jsonrpcserver import Methods
Methods(ping=lambda:'pong').serve_forever()
Using the @add
decorator:
from jsonrpcserver import methods
@methods.add
def ping():
return 'pong'
if __name__ == '__main__':
methods.serve_forever()
Socket.IO¶
$ pip install jsonrpcserver flask flask-socketio eventlet
from flask import Flask
from flask_socketio import SocketIO, send
from jsonrpcserver import methods
from jsonrpcserver.response import NotificationResponse
app = Flask(__name__)
socketio = SocketIO(app)
@methods.add
def ping():
return 'pong'
@socketio.on('message')
def handle_message(request):
response = methods.dispatch(request)
if not response.is_notification:
send(response, json=True)
if __name__ == '__main__':
socketio.run(app, port=5000)
See blog post.
Tornado¶
$ pip install jsonrpcserver tornado
from tornado import ioloop, web
from jsonrpcserver.aio import methods
from jsonrpcserver.response import NotificationResponse
@methods.add
async def ping():
return 'pong'
class MainHandler(web.RequestHandler):
async def post(self):
request = self.request.body.decode()
response = await methods.dispatch(request)
if not response.is_notification:
self.write(str(response))
app = web.Application([(r"/", MainHandler)])
if __name__ == '__main__':
app.listen(5000)
ioloop.IOLoop.current().start()
See blog post.
Websockets¶
$ pip install jsonrpcserver websockets
import asyncio
import websockets
from jsonrpcserver.aio import methods
from jsonrpcserver.response import NotificationResponse
@methods.add
async def ping():
return 'pong'
async def main(websocket, path):
request = await websocket.recv()
response = await methods.dispatch(request)
if not response.is_notification:
await websocket.send(str(response))
start_server = websockets.serve(main, 'localhost', 5000)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
See blog post.
Werkzeug¶
$ pip install jsonrpcserver werkzeug
from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple
from jsonrpcserver import methods
@methods.add
def ping():
return 'pong'
@Request.application
def application(request):
r = methods.dispatch(request.data.decode())
return Response(str(r), r.http_status, mimetype='application/json')
if __name__ == '__main__':
run_simple('localhost', 5000, application)
See blog post.
ZeroMQ¶
$ pip install jsonrpcserver pyzmq
import zmq
from jsonrpcserver import methods
from jsonrpcserver.response import NotificationResponse
socket = zmq.Context().socket(zmq.REP)
@methods.add
def ping():
return 'pong'
if __name__ == '__main__':
socket.bind('tcp://*:5000')
while True:
request = socket.recv().decode()
response = methods.dispatch(request)
socket.send_string(str(response))
See blog post.
ZeroMQ (asynchronous)¶
$ pip install jsonrpcserver aiozmq
import asyncio
import aiozmq
import zmq
from jsonrpcserver.aio import methods
from jsonrpcserver.response import NotificationResponse
@methods.add
async def ping():
return 'pong'
async def main():
rep = await aiozmq.create_zmq_stream(zmq.REP, bind='tcp://*:5000')
while True:
request = await rep.read()
response = await methods.dispatch(request[0].decode())
rep.write((str(response).encode(),))
if __name__ == '__main__':
asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
asyncio.get_event_loop().run_until_complete(main())
See blog post.