Writing a minimal JSON Webserver using http.server
Introduction
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.
Structure
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)))
self.end_headers()
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.