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 import method, async_dispatch as dispatch


@method
async def ping():
    return "pong"


async def handle(request):
    request = await request.text()
    response = await dispatch(request)
    if response.wanted:
        return web.json_response(response.deserialized(), status=response.http_status)
    else:
        return web.Response()


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 method, dispatch


@method
def ping():
    return "pong"


@csrf_exempt
def jsonrpc(request):
    response = dispatch(request.body.decode())
    return JsonResponse(
        response.deserialized(), status=response.http_status, safe=False
    )

See blog post.

Flask

$ pip install jsonrpcserver flask

from flask import Flask, request, Response
from jsonrpcserver import method, dispatch

app = Flask(__name__)


@method
def ping():
    return "pong"


@app.route("/", methods=["POST"])
def index():
    response = dispatch(request.get_data().decode())
    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

"""Using Python's built-in HTTPServer"""
from http.server import BaseHTTPRequestHandler, HTTPServer
from jsonrpcserver import method, dispatch


@method
def ping():
    return "pong"


class TestHttpServer(BaseHTTPRequestHandler):
    def do_POST(self):
        # Process request
        request = self.rfile.read(int(self.headers["Content-Length"])).decode()
        response = 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 method.

$ pip install jsonrpcserver

The quickest way to serve a method:

from jsonrpcserver import method
method(ping=lambda:'pong')
serve()

Using the @method decorator:

"""Using jsonrpcserver's built-in serve() function"""
from jsonrpcserver import method, serve


@method
def ping():
    return "pong"


if __name__ == "__main__":
    serve()

Socket.IO

$ pip install jsonrpcserver flask flask-socketio eventlet

from flask import Flask
from flask_socketio import SocketIO, send
from jsonrpcserver import method, dispatch

app = Flask(__name__)
socketio = SocketIO(app)


@method
def ping():
    return "pong"


@socketio.on("message")
def handle_message(request):
    response = dispatch(request)
    if response.wanted:
        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 import method, async_dispatch as dispatch


@method
async def ping() -> str:
    return "pong"


class MainHandler(web.RequestHandler):
    async def post(self) -> None:
        request = self.request.body.decode()
        response = await dispatch(request)
        if response.wanted:
            self.write(str(response))


app = web.Application([(r"/", MainHandler)])

if __name__ == "__main__":
    app.listen(5000)
    ioloop.IOLoop.current().start()

See blog post.

Vibora

$ pip install jsonrpcserver vibora

from vibora import Vibora, Request
from vibora.responses import JsonResponse
from jsonrpcserver import method, async_dispatch as dispatch

app = Vibora()


@method
async def ping():
    return "pong"


@app.route("/", methods=["POST"])
async def home(request: Request):
    request = await request.stream.read()
    response = await dispatch(request.decode())
    return JsonResponse(response.deserialized())


if __name__ == "__main__":
    app.run()

See blog post.

Websockets

$ pip install jsonrpcserver websockets

import asyncio
import websockets
from jsonrpcserver import method, async_dispatch as dispatch


@method
async def ping():
    return "pong"


async def main(websocket, path):
    response = await dispatch(await websocket.recv())
    if response.wanted:
        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 method, dispatch


@method
def ping():
    return "pong"


@Request.application
def application(request):
    response = dispatch(request.data.decode())
    return Response(str(response), response.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 method, dispatch

socket = zmq.Context().socket(zmq.REP)


@method
def ping():
    return "pong"


if __name__ == "__main__":
    socket.bind("tcp://*:5000")
    while True:
        request = socket.recv().decode()
        response = dispatch(request)
        socket.send_string(str(response))

See blog post.

ZeroMQ (asynchronous)

$ pip install jsonrpcserver aiozmq

import asyncio
import aiozmq
import zmq
from jsonrpcserver import method, async_dispatch as dispatch


@method
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 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.