Writing a minimal JSON Webserver using http.server


Recently I wrote a small API webservice with the explicit goal of just using just python's built in http.server.

It wasn't very straight forward, as it is really not designed to be a proper web server, but mostly used to serve a single directory, but with a bit of work, it does the job.


The web server is implemented with a generic class subclassing =BaseHTTPRequestHandler=, it looks up paths in a dictionary, and calls a class per URL. This class implements the basic GET/POST etc. verbs as functions. If these functions return any non-string object, these objects will be serialized as JSON.

Base class

class BaseJSONHandler(BaseHTTPRequestHandler): """Generic handler of HTTP requests"""

def __init__(self, *args, urlmap=None, **kwargs): self.urlmap = {} if urlmap is not None: self.urlmap = urlmap

super().__init__(*args, **kwargs)

def _request(self): data = None length = int(self.headers.get('content-length', 0)) if length > 0: data = self.rfile.read(length) request = { "server": self.server, "command": self.command, "rfile": self.rfile, "wfile": self.wfile, "data": data, "headers": self.headers } return request

def send_error(self, code, message=None, explain=None): """JSON style override for send_error"""

try: shortmsg, longmsg = self.responses[code] except KeyError: shortmsg, longmsg = '???', '???'

if message is None: message = shortmsg if explain is None: explain = longmsg

self.log_error("code %d, message %s", code, message) self.send_response(code, message) self.send_header('Connection', 'close')

body = None

if (code >= 200 and code not in ( HTTPStatus.NO_CONTENT, HTTPStatus.RESET_CONTENT, HTTPStatus.NOT_MODIFIED )):

timestamp = datetime.datetime.utcnow().\ replace(tzinfo=datetime.timezone.utc).isoformat() content = { "timestamp": timestamp, "status": code, "error": message, "message": explain, "path": self.path } body = json.dumps(content).encode('utf-8') self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body)))


if self.command != 'HEAD' and body: self.wfile.write(body)

def _set_headers(self): self.send_response(HTTPStatus.OK.value) self.send_header('Content-type', 'application/json') # Send CORS for good measures self.send_header('Access-Control-Allow-Origin', '*') self.end_headers()

def handle_one_request(self): """Handle a single HTTP request.

This method is overridden to better reuse code"""

try: # pylint: disable=attribute-defined-outside-init self.raw_requestline = self.rfile.readline(65537) if len(self.raw_requestline) > 65536: # pylint: disable=attribute-defined-outside-init self.requestline = '' self.request_version = '' self.command = '' self.send_error(HTTPStatus.REQUEST_URI_TOO_LONG) return if not self.raw_requestline: self.close_connection = True return if not self.parse_request(): return

method = self.command

# Some method should stay local to this file, like OPTIONS and HEAD mname = 'do_' + self.command if hasattr(self, mname): method = getattr(self, mname) method() self.wfile.flush() else: self.resolve_request() self.wfile.flush()

except socket.timeout as err: # pylint: disable=attribute-defined-outside-init self.log_error("Request timed out: %r", err) self.close_connection = True

def resolve_request(self): """Pass on request to a class""" method = self.command url = urlsplit(self.path) path = url.path mname = 'do_' + self.command

if path in self.urlmap: cls = self.urlmap[path](self) request = self._request() request['query'] = parse_qs(url.query) if hasattr(cls, mname): method = getattr(cls, mname) try: response = method(request) self._set_headers() if response is not None: if isinstance(response, str): self.wfile.write(response.encode('utf-8')) else: self.wfile.write( json.dumps(response).encode('utf-8') ) except BadRequestException as err: self.send_error( 400, message=err.message, explain=err.explain ) except MethodNotAllowed as err: self.send_error( 405, message=err.message, explain=err.explain )

# pylint: disable=broad-except except Exception as ex: self.send_error( 500, message="Unexpected error", explain=ex )

else: self.send_error( 405, message="Method Not Allowed", explain=f"Request method '{method}' not allowed" ) else: self.send_error( 404, message="Not Found", explain="No message available" )

def do_OPTIONS(self): """ Send CORS headers on OPTIONS """ # pylint: disable=invalid-name self.send_response(HTTPStatus.NO_CONTENT.value) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST') self.send_header('Access-Control-Allow-Headers', 'content-type') self.end_headers()

The parent class uses the handle_one_request method to call a method for each HTTP Verb, like do_GET() or do_POST(). I'm overriding this to look up the path in a dictionary, and instead call the method there.

I'm also implementing a number of shorthand Exception classes that all serve to format my exceptions a certain way. This is because I'm using this to mock another system, and I want my exceptions to be identical to that.

There's no real magic to the child classes responsible for serving the URLs. They are all called with a request dictionary, containing various properties of the HTTP request, such as the query strings, and the payload, if any.

This allows you to specify a class like this:

class HelloWorldController: def post(self, request): thing = request['query'].get('thing', [])[0] if len(thing) == 0: thing = "World"

return { "greeting": "Hello", "thing": thing }

This can then be instantiated like this:

urlmap = { "/greet": HelloWorldController } handler = partial(BaseJSONHandler, urlmap=urlmap) httpd = HTTPServer(('', 8080), handler) httpd.serve_forever()

simple and sweet.

With this class based approach, I'm basically creating infinite extensibility, but I'm sure the bottle neck in http.server will be apparent if you try to do too much with this.

comments powered by Disqus