Skip to content

Instantly share code, notes, and snippets.

@jamesgregson
Created January 4, 2020 19:05
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jamesgregson/5e0db7795ca60ae1c61030d63a676533 to your computer and use it in GitHub Desktop.
Save jamesgregson/5e0db7795ca60ae1c61030d63a676533 to your computer and use it in GitHub Desktop.
(Over) Simplified JSONRPC in Python
'''Very simplified JSON-RPC server with endpoint registration
Try with:
curl -X POST -H "Content-type: application/json" -d '{"jsonrpc": "2.0", "id": 1, "method": "say_hello", "params": {"name": "James", "age": 11}}' http://127.0.0.1:8000
curl -X POST -H "Content-type: application/json" -d '{"jsonrpc": "2.0", "id": 1, "method": "say_hello", "params": {"name": "James"}}' http://127.0.0.1:8000
curl -X POST -H "Content-type: application/json" -d '{"jsonrpc": "2.0", "id": 1, "method": "say_hello", "params": ["James", 10]}' http://127.0.0.1:8000
curl -X POST -H "Content-type: application/json" -d '{"jsonrpc": "2.0", "id": 1, "method": "say_hello", "params": ["James"]}' http://127.0.0.1:8000
'''
import json
import socketserver
from http import server
class RPCHandler( server.BaseHTTPRequestHandler ):
endpoints = {}
@staticmethod
def register( func ):
RPCHandler.endpoints[func.__name__] = func
return func
def do_POST( self ):
try:
# check that constent type is 'application/json'
con_type = self.headers['Content-type'].split(';')[0]
if con_type == 'application/json':
# get the posted data and parse the request object
con_len = int(self.headers['Content-length'])
req = json.loads( self.rfile.read(con_len).decode('utf-8') )
# handle the request if registered
if req['method'] in RPCHandler.endpoints:
if isinstance(req['params'],dict): result = RPCHandler.endpoints[req['method']]( **req['params'] )
else: result = RPCHandler.endpoints[req['method']]( *req['params'] )
resp = json.dumps({
'jsonrpc': '2.0',
'method': req['method'],
'id': req['id'],
'result': result
}).encode('utf-8')
# everything after here unlikely to raise exceptions..
self.send_response( 200 )
self.send_header('Content-type','application/json')
self.end_headers()
self.wfile.write( resp )
return
except: pass
# on exception, malformed input or invalid method just send back 501
self.send_error(501,'Malformed or invalid request, check method name and arguments!')
@RPCHandler.register
def say_hello( name, age=6 ):
return 'Hello {}, aged {}'.format(name,age)
if __name__ == '__main__':
address = ('',8000)
server = server.HTTPServer( address, RPCHandler )
try:
print('Server up on port {}'.format(address[1]))
server.serve_forever()
except KeyboardInterrupt:
pass
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment