jsonrpcserver

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.