Adding routes into my bodged-together web server
Kyle Martineau-McFarlane
18 February 2023
Introduction
In the last part, I built a very simple, very shoddy and very basic web server. At the moment, it doesn't pay any attention to the path of the request, and just responds to everything with a "Hello, world" message. In this part, I'll be adding in a request-handling framework that will allow us to do something like this to add in a path-based route to return a status, dict of headers, and body:
@server.register("GET", "/help")
def help(request):
return (200, {}, "<p>Try turning it off, and then turning it on.</p>")
Handling the request
An HTTP request looks something like this:
GET /some-path/ HTTP/1.1 Host: localhost:8080 User-Agent: curl/7.74.0 Accept: */*
The first line stipulates the method (GET), the path (/some-path/) and the version of HTTP being used. The next lines (until a double-newline) are headers, and then if any content is provided (as in a POST, PATCH or PUT), it will be added at the end. At the moment, I'm mostly interested in GETs. We will need a function to parse the first line of the request to identify the path, and a map to store the known paths. We then need the decorator to add to functions and a dispatcher to call the appropriate functions.
To build our parsing function, we need to work through the lines of the request string in order. The first one is straightforward: we can split it on the space. We then collect a dict of headers, which ends with the double new line, before we get the content. We'll need some logic to handle this; if we encounter a blank line, is it the second one? If so, the next line will start the content section. If it's the first, we need to flag it as such so that the next blank line can trigger the content section. This function should handle it:
def parse_request(self, request):
request_lines = request.split("\r\n")
method, path, _ = request_lines[0].split(" ")
headers = {}
content = []
blank_lines = 0
for i in range(1, len(request_lines)):
if request_lines[i] != "":
if blank_lines == 0:
# header section
key, value = request_lines[i].split(": ")
headers[key.lower()] = value
if blank_lines == 2:
content.append(request_lines[i])
else:
if blank_lines == 0:
blank_lines = 2
else:
blank_lines = 1
return {
"method": method,
"path": path,
"headers": headers,
"content": "\n".join(content),
}
This will parse our request into a useful dictionary containing all of the parts of our request. The algorithm is fairly straightforward and parses the text in a fairly obvious line-by-line approach.
We now need a map to link paths and methods to the functions that we will use to serve those paths. We can do this with a simple map which we add into our __init__ function:
def __init__(self, address, port):
self.address = address
self.port = port
self.paths = {}
To populate this, we need our decorator function. It's going to look like this:
def register(self, *args, **kwargs):
method = args[0]
path = args[1]
def inner(func):
self.paths[f"{method}__{path}"] = func
return inner
In Python, a decorator is a function (register) which wraps another function (func) returning a third function (inner), which replaces the original function func. This inner function will perform some other operations before func is called; in this case, rather than it being called directly, we are storing the function func in our paths map so that it can be called later without needing to worry about the actual name of the function.
Dispatching
We now are nearly there. The final bit of complication is to work out which decorated function to call, and call it:
request = self.parse_request(request)
method = request["method"]
path = request["path"]
func = self.paths.get(f"{method}__{path}")
if func:
response = func(request)
else:
pass # TODO handle 404
We'll sort out what we do with 404s next time. For now, we'll just make sure our test requests are valid paths.
We now need another utility function to convert the output of our help() function into a valid HTTP response. We'll call it render_response:
def render_response(self, response):
status, headers, response_body = response
header_string = "\r\n".join([f"{k}: {v}" for k, v in headers.items()])
return f"HTTP/1.0 {status}\r\n{header_string}\r\n\r\n{response_body}"
We can insert this instead of the message response we had and now we have a functioning HTTP server!
$ curl http://localhost:8080/help <p>Try turning it off, and then turning it on.</p>
It works! For reference, here's the current script (including some useful traceback information):
import socket
import traceback
class Server:
def __init__(self, address, port):
self.address = address
self.port = port
self.paths = {}
def register(self, *args, **kwargs):
method = args[0]
path = args[1]
def inner(func):
self.paths[f"{method}__{path}"] = func
return inner
def parse_request(self, request):
request_lines = request.split("\r\n")
method, path, _ = request_lines[0].split(" ")
headers = {}
content = []
blank_lines = 0
for i in range(1, len(request_lines)):
if request_lines[i] != "":
if blank_lines == 0:
# header section
key, value = request_lines[i].split(": ")
headers[key.lower()] = value
if blank_lines == 2:
content.append(request_lines[i])
else:
if blank_lines == 0:
blank_lines = 2
else:
blank_lines = 1
return {
"method": method,
"path": path,
"headers": headers,
"content": "\n".join(content),
}
def render_response(self, response):
status, headers, response_body = response
header_string = "\r\n".join([f"{k}: {v}" for k, v in headers.items()])
return f"HTTP/1.0 {status}\r\n{header_string}\r\n\r\n{response_body}"
def serve(self):
self.s = socket.socket()
self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
addr = socket.getaddrinfo(self.address, self.port)[0][-1]
self.s.bind(addr)
self.s.listen(5)
print(f"Listening to {self.address}:{self.port}")
message = (
"HTTP/1.0 200\r\nContent-Type: text/plain\r\n\r\nHello, world.".encode(
"utf-8"
)
)
while True:
try:
self.cl, addr = self.s.accept()
with self.cl as cl:
request = cl.recv(1024)
request = request.decode("utf-8")
request = self.parse_request(request)
method = request["method"]
path = request["path"]
func = self.paths.get(f"{method}__{path}")
if func:
response = func(request)
else:
pass # TODO handle 404
cl.send(self.render_response(response).encode("utf-8"))
except Exception as e:
print(traceback.format_exc())
print(e)
self.s.close()
raise SystemExit(1)
server = Server("0.0.0.0", 8080)
@server.register("GET", "/help")
def help(request):
return (200, {}, "<p>Try turning it off, and then turning it on.</p>")
if __name__ == "__main__":
server.serve()
In the next instalment, we'll refine the code by adding some error handlers and a templating "engine".