Creating a Web Framework with Python

Yes, let’s create yet another web framework, because why not? By the end of this post, we’ll be able to use our new framework as such:

Open Table of contents

Introduction
WSGI
ASGI
The Web Framework
Application API
Improving the Request and Response Objects
Routing
Handling Routes
Putting it all together

Introduction

First of all, we need to understand how to talk to a web server from our Python web framework. In the early days of Python web development, there were many frameworks out there, but they had limited compatibility across different web servers. To solve that, they came up with wsgi which stands for Web Server Gateway Interface¹. Now web servers have a convenient and standard way to talk to each other, also known as an interface.

WSGI

The Web Server Gateway Interface (WSGI) is a simple calling convention for web servers to forward requests to web applications or frameworks written in the Python¹.

A WSGI-compatible web server expects to receive a callable that takes two arguments. Our callable is application . The first argument is a dict , that holds the information about the incoming request. The second is another Callable responsible for setting the response code and response headers.

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    yield b'Hello, World!\n'

I picked the gunicorn server to serve our app. The entry-point expects the callable as below:

$ gunicorn <module>:<callable>

From the command-line, we can start the app.

$ gunicorn wsgi:application
[INFO] Starting gunicorn 20.0.4
[INFO] Listening at: http://127.0.0.1:8000 (35567)
[INFO] Using worker: sync
[INFO] Booting worker with pid: 35570

Now our web app is running on port 8000 so we can access it via browser or curl:

$ curl http://127.0.0.1:8000/
Hello, World!

ASGI

ASGI is a spiritual successor to WSGI, the long-standing Python standard for compatibility between web servers, frameworks, and applications.

You might have noticed the async keyword in the snippet at the beginning. We are creating an async — compatible framework, and for that, we need to learn another calling convention. But since we covered the WSGI standard above, that should give us a head start.

An ASGI-compatible web server expects to receive an async callable , which basically means a function declared with the async keyword. Our async callable is application . The first argument is also a dict that holds information about the incoming request. The second is another async callable responsible for receiving events from the client. The third is also an-async callable responsible for sending messages to the client.

async def application(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            ['content-type', 'text/plain'],
        ],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!\n',
    })

This time I picked uvicorn to serve our async app. So as we did previously, starting the app is as simple as:

$ uvicorn asgi:application
INFO: Started server process [36724]
INFO: Waiting for application startup.
INFO: ASGI 'lifespan' protocol appears unsupported.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

We can now head to our localhost and access the app running on port 8000.

$ curl http://127.0.0.1:8000/
Hello, world!

It’s very fair to say that the web servers are responsible for a lot of the heavy lifting, which is a good thing. It leaves us with a lot of flexibility.

The Web Framework

Now that we know how servers and apps talk to each other, we can design our own web framework and make architectural decisions.

Application API

Since the webserver is always expecting to receive a callable, we can take advantage of Python’s magic method __call__ and add it to our API. By implementing this, we are basically telling Python that this object can be called, and when that happens, it will follow the code within that method. And as we’ve seen before, the call method needs to receive three arguments.

import uvicorn


class Pullo:
    async def __call__(self, scope, receive, send):
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    ["content-type", "text/plain"],
                ],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": b"Hello, world!\n",
            }
        )


app = Pullo()


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

Notice that now we are passing the callable to uvicorn directly within python with uvicorn.run(app, **kwargs). This is the same as if you run $ uvicorn pullo:app from the command-line. The former is good for development and debugging, so you don’t have to type a lot of things all the time from the command-line. You can just run $ python pullo.py.

Improving the Request and Response Objects

Writing down those response dictionaries and calling send can be quite boring, and it clutters our __call__ method a bit. Since we are focusing on the web framework, we can outsource parsing requests and responses to a third-party library called Starlette.

import uvicorn
from starlette.responses import Response


class Pullo:
    async def __call__(self, scope, receive, send):
        response = Response("Hello, world!\n")
        await response(scope, receive, send)


app = Pullo()


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

Our call method looks much cleaner now. The Response class takes a few other arguments such as status_code , media_type, and others, so we can use that in the future for more flexibility. It will return an async callable as well. After building the response object, we can just await the callable response object passing back the arguments.

Routing

So far, we’ve been able only to serve requests coming to our root / path. If you’re familiar with other Python web frameworks like Flask or FastAPI, which are amazing, by the way, the following syntax will be quite straightforward to grasp.

Now we need to build a route builder, so by decorating a function with @app.route("<path>"), our handler, later on, knows where to match the path of the incoming request and the function responsible for that route.

class Pullo:
    def __init__(self):
        self.routes = {}

    def route(self, path):
        def wrapper(handler):
            self.routes[path] = handler
            return handler

        return wrapper
    ...


app = Pullo()


@app.route("/hello")
async def hello():
    return "Saying hi from hello."

We added an attribute to our class called routes , where we’ll store a dictionary with the mapping of path and function. We also added a method called route responsible for the mapping. Below we can see how the routes dictionary looks like after decorating a function with the route decorator.

>>> print(app.routes)
{'/hello': <function hello at 0x10a702620>}

Handling Routes

Our route decorator can now map the paths with their respective endpoints. So now, we need to handle incoming requests and point them to the correct endpoint. We’ll do that from within our __call__ method.

import uvicorn
from starlette.requests import Request
from starlette.responses import Response


class Pullo:
    ...

    async def handler(self, request):
        endpoint = self.routes.get(request.url.path)

        if endpoint:
            output = await endpoint()
            return Response(output)

        return Response("Not Found!", status_code=404)

    async def __call__(self, scope, receive, send):
        request = Request(scope)
        response = await self.handler(request)

        await response(scope, receive, send)

...

As soon as a request comes, we parse its content and pass that object to the handler method. The handler then fetches the correct endpoint from the routes dictionary. If there is a matching route, we return the endpoint. Else we create a new response object with a not found message and status code. When there is a match, we also need to evaluate the endpoint to perform the business logic within that function and then create a response object with the output.

Back to the __call__ method, we can finally await the response object and send the content back to the webserver.

That’s it. Our tiny web framework can correctly route incoming requests and perform all sorts of operations to send it back to the webserver.

Putting it all together

import uvicorn
from starlette.requests import Request
from starlette.responses import Response


class Pullo:
    def __init__(self):
        self.routes = {}

    def route(self, path):
        def wrapper(handler):
            self.routes[path] = handler
            return handler

        return wrapper

    async def handler(self, request):
        endpoint = self.routes.get(request.url.path)

        if endpoint:
            output = await endpoint()
            return Response(output)

        return Response("Not Found!", status_code=404)

    async def __call__(self, scope, receive, send):
        request = Request(scope)
        response = await self.handler(request)

        await response(scope, receive, send)


app = Pullo()


@app.route("/hello")
async def hello():
    return "Saying hi from hello."


@app.route("/")
async def home():
    return "Saying hi from home."


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

$ curl http://127.0.0.1:8000/hello
Saying hi from hello.
$ curl http://127.0.0.1:8000/      
Saying hi from home.

Thank you for sticking around. Please do let me know if you encounter some mistakes, typos, or would like to suggest something, etc. Feedback is always welcomed.

Originally posted at alephmelo.medium.com in 2020.

Pullo is Finnish for Flask/Bottle.