ASGI from scratch - Let's build an ASGI web framework
Tags: asgi, python
Published: 2020-06-21 04:34:07

---
title: ASGI from scratch - Let's build an ASGI web framework
published: true
description: build an ASGI web framework from scratch
tags: ASGI,Python
canonical_url: https://shenli.dev/2020/06/20/asgi-from-scratch.html
//cover_image: https://direct_url_to_image.jpg
---
# Intro
The first time I used [ASGI][asgi](_Asynchronous Server Gateway Interface_) was through [Channels][channels] 1.0 when ASGI spec was still a draft. It was my first interview project which helped me get my current job. It felt magical at that time how easy it is to add WebSocket functionality to my Django app and handles authentication and other Django related things for me seamlessly.

ASGI specification is now at version 3 at the time of writing and both ASGI and Channels became part of Django Software Foundation. Compared to the draft version, it has matured a lot with added lifecycle calls and better application format, etc. Most excitingly, a healthy and fast-growing community is forming and we are seeing more and more ASGI servers running in production environments. At my company, we are serving a few million requests per day through ASGI running on [Daphne][daphne], Netflix's [Dispatch][dispatch] is based on [FastAPI][fastapi], a popular ASGI web application framework, and apparently, Microsoft is [using it][fast-api-pr] too.

I would humbly advise anyone building web services in Python to learn about ASGI. And the best way to learn something is to built things with it, so in this blog post, I'll walk through the steps to build a micro web application framework that speaks ASGI. I hope it can help explain how ASGI works.

# Settings the goal
Before writing the first line of code, we need to have a basic understanding of what ASGI is and what we are building towards. 
## How ASGI works
Here's a simple diagram showing how ASGI works at a high level.
![basic-asgi-graph](https://dev-to-uploads.s3.amazonaws.com/i/jkl5d9lsfokr05aggpb9.png)
To put it in simple words, A browser(client), establishes a connection to ASGI server with a certain type of request (HTTP or WebSocket), the ASGI server then calls ASGI  application with information about the connection, encapsulated in a python dictionary called `scope`, and two callbacks, named `send` and `receive`, that the application can use to send and receive messages between server and client.

Here's an example HTTP request scope
```python
{
    "type": "http",
    "http_version": "1.1",
    "server": ("127.0.0.1", 8000),
    "client": ("127.0.0.1", 60457),
    "scheme": "http",
    "method": "GET",
    "root_path": "",
    "path": "/hello/a",
    "raw_path": b"/hello/a",
    "query_string": b"",
    "headers": [
        (b"host", b"localhost:8000"),
        (b"connection", b"keep-alive"),
        (
            b"user-agent",
            b"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Safari/537.36",
        ),
        (
            b"accept",
            b"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
        ),
        (b"accept-encoding", b"gzip, deflate, br"),
        (b"accept-language", b"en-US,en;q=0.9"),
        (
            b"cookie",
            b'csrftoken=dDA2IAPrvgPc7hkyBSyctxDk78KmhHAzUqR0LUpjXI3Xgki0QrGEWazE3RGZuLGl',
        ),
    ],
}
```

You might notice that `scope` is not too different from a WSGI `environ`. In fact, ASGI interface is very similar to WSGI interface, but instead of getting a `environ` and `start_response` to send headers and using the return value of WSGI application as the response body, ASGI interfaces with the connection and allows us to receive and send messages multiple times during the lifecycle of the connection __asynchronously__ until the connection is closed.  This allows a nice interface for both WebSocket and HTTP. 

It's also totally possible to wrap a WSGI application inside an ASGI application, just prepare a WSGI `environ` and `start_response` based on `scope`, `receive`, and `send` then call the WSGI application and it would work. If you delegate that call into a thread pool or something similar, you just made your WSGI application asynchronous. This is roughly how Channels wraps around Django.

## Define ASGI framework
When I say ASGI framework I refer it as a framework that makes building ASGI application easier and this does not include the ASGI server part. I'm mentioning this because some of the earlier Python asynchronous web frameworks have their own server implementation that also takes over tasks such as parsing  HTTP requests, handles network connections, etc. We are not doing those in ASGI web framework. As a spiritual successor to WSGI, where web servers, such as Gunicorn and uwsgi, and web frameworks, such as Flask and Django, are separated, ASGI has this separation too.

So, what does an ASGI application look like? 

### ASGI Hello World
A simple ASGI hello world application can be written as:
```python
async def application(scope, receive, send):
    name = scope["path"].split("/", 1)[-1] or "world"
    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [[b"content-type", b"text/plain"],],
        }
    )
    await send(
        {
            "type": "http.response.body",
            "body": f"Hello, {name}!".encode(),
            "more_body": False,
        }
    )
```
`http.response.start` starts an HTTP response sending status code and response headers. In this example, it responds with the 200 OK status code and  has `content-type` set to `text/plain` in the headers.  `http.response.body` sends the response body, the `more_body` key tells the server if the response is finished. ASGI server might use this to know if a connection should be closed or automatically decide between a `content-length` header or a chunked encoding.

We can run the application with [uvicorn][uvicorn]:
```bash
uvicorn asgi-hello:application
```
And you should be able to visit `http://localhost:8000/` and get `Hello, world`.Visiting `http://localhost:8000/tom` would get you `Hello, tom`.  

> By the way, uvicorn is pretty fast, a simple benchmark with `wrk -d10s http://localhost:8000/hi` on a 2018 lowest spec MacBook Air yields `Requests/sec:  27857.87`.

Although this approach works with a simple hello world example, it's not exactly convenient to write a more complex application this way. For one, it doesn't do routing, if you want to respond differently for different paths, you'll probably end up with a huge  `if ... else if ... else` clause. Secondly, having to write the ASGI message every time in the form of a python dict is quite arduous. Third, in a complex application, it gets harder to track the status of the connection, such as is the response started, is the response ended, should I start the response here, etc.

### Goal
With the new framework, I hope to be able to write an ASGI application like this:
```python
import asyncio
from aaf import aaf # Another ASGI framework
from aaf.routing import Router
from aaf.response import HttpResponse

router = Router()

@router.route('/')
@router.route('/<name>')
async def hello(connection, name='world'):
	return HttpResponse(f"Hello, {name}")


@router.route('/count')
@router.route('/count/<int:number>')
async def count(connection, number=10):
	for i in range(number):
		await connection.send(f'count {i}\n', finish=False)
		await asyncio.sleep(1)
	await connection.send('', finish=True)


@router.route('/echo')
async def echo(connection):
	body = await connection.body()
	await connection.send(body, finish=True)


app = aaf([router])
```
I hope this snippet of how I want the framework to look like is self-explanatory. But here are some of the key things I want to achieve: 
1. It should be able to handle HTTP response declaratively and imperatively.
1. It should support Flask style routing with parameter parsing.

# Building the framework
## Connection class
The `Connection` class will represent an ASGI HTTP or WebSocket connection. It's a class that encapsulates the three basic elements in ASGI, namely `scope`, `send` and `receive`, and expose some convenient methods and properties so that users don't need to verbosely write out all the ASGI messages and parse everything, such as cookies and headers, from `scope`. But it should allow users to access the original `scope`, `send` and `receive` when they want to, so that the composability of ASGI applications is maintained. For example, it should allow user to delegate certain `connection`s to another ASGI application by calling `another_asgi_app(connection.scope, connectionn.asgi_send, connection.asgi_receive)`. 

Here's a simple implementation of the `Connection` class.
```python
from enum import Enum
from functools import cached_property
from http.cookies import SimpleCookie
from typing import Any, Awaitable, Callable, Optional, Union
from urllib.parse import parse_qsl, unquote_plus

from werkzeug.datastructures import Headers, MultiDict

CoroutineFunction = Callable[[Any], Awaitable]


class ConnectionType(Enum):
    HTTP = "HTTP"
    WebSocket = "WebSocket"


class Connection:
    def __init__(
        self, scope: dict, *, send: CoroutineFunction, receive: CoroutineFunction
    ):
        self.scope = scope
        self.asgi_send = send
        self.asgi_receive = receive

        self.started = False
        self.finished = False
        self.resp_headers = Headers()
        self.resp_cookies: SimpleCookie = SimpleCookie()
        self.resp_status_code: Optional[int] = None

        self.http_body = b""
        self.http_has_more_body = True
        self.http_received_body_length = 0

    @cached_property
    def req_headers(self) -> Headers:
        headers = Headers()
        for (k, v) in self.scope["headers"]:
            headers.add(k.decode("ascii"), v.decode("ascii"))
        return headers

    @cached_property
    def req_cookies(self) -> SimpleCookie:
        cookie = SimpleCookie()
        cookie.load(self.req_headers.get("cookie", {}))
        return cookie

    @cached_property
    def type(self) -> ConnectionType:
        return (
            ConnectionType.WebSocket
            if self.scope.get("type") == "websocket"
            else ConnectionType.HTTP
        )

    @cached_property
    def method(self) -> str:
        return self.scope["method"]

    @cached_property
    def path(self) -> str:
        return self.scope["path"]

    @cached_property
    def query(self) -> MultiDict:
        return MultiDict(parse_qsl(unquote_plus(self.scope["query_string"].decode())))

    async def send(self, data: Union[bytes, str] = b"", finish: Optional[bool] = False):
        if self.finished:
            raise ValueError("No message can be sent when connection closed")
        if self.type == ConnectionType.HTTP:
            if isinstance(data, str):
                data = data.encode()
            await self._http_send(data, finish=finish)
        else:
            raise NotImplementedError()

    async def _http_send(self, data: bytes = b"", *, finish: bool = False):
        if not self.started:
            if finish:
                self.put_resp_header("content-length", str(len(data)))
            await self.start_resp()
        await self.asgi_send(
            {"type": "http.response.body", "body": data or b"", "more_body": True}
        )
        if finish:
            await self.finish()

    async def finish(self, close_code: Optional[int] = 1000):
        if self.type == ConnectionType.HTTP:
            if self.finished:
                raise ValueError("Connection already finished")
            if not self.started:
                self.resp_status_code = 204
                await self.start_resp()
            await self.asgi_send(
                {"type": "http.response.body", "body": b"", "more_body": False}
            )
        else:
            raise NotImplementedError()
            # await self.asgi_send({"type": "websocket.close", "code": close_code})
        self.finished = True

    async def start_resp(self):
        if self.started:
            raise ValueError("resp already started")
        if not self.resp_status_code:
            self.resp_status_code = 200
        headers = [
            [k.encode("ascii"), v.encode("ascii")] for k, v in self.resp_headers.items()
        ]
        for value in self.resp_cookies.values():
            headers.append([b"Set-Cookie", value.OutputString().encode("ascii")])
        await self.asgi_send(
            {
                "type": "http.response.start",
                "status": self.resp_status_code,
                "headers": headers,
            }
        )
        self.started = True

    async def body_iter(self):
        if not self.type == ConnectionType.HTTP:
            raise ValueError("connection type is not HTTP")
        if self.http_received_body_length > 0 and self.http_has_more_body:
            raise ValueError("body iter is already started and is not finished")
        if self.http_received_body_length > 0 and not self.http_has_more_body:
            yield self.http_body
        req_body_length = (
            int(self.req_headers.get("content-length", "0"))
            if not self.req_headers.get("transfer-encoding") == "chunked"
            else None
        )
        while self.http_has_more_body:
            if req_body_length and self.http_received_body_length > req_body_length:
                raise ValueError("body is longer than declared")
            message = await self.asgi_receive()
            message_type = message.get("type")
            if message.get("type") == "http.disconnect":
                raise ValueError("Disconnected")
            if message_type != "http.request":
                continue
            chunk = message.get("body", b"")
            if not isinstance(chunk, bytes):
                raise ValueError("Chunk is not bytes")
            self.http_body += chunk
            self.http_has_more_body = message.get("more_body", False) or False
            self.http_received_body_length += len(chunk)
            yield chunk

    async def body(self):
        return b"".join([chunks async for chunks in self.body_iter()])

    def put_resp_header(self, key, value):
        self.resp_headers.add(key, value)

    def put_resp_cookie(self, key, value):
        self.resp_cookies[key] = value
```
I hope this code is easy to read. All it does is providing us with an interface that makes it easier to access info about the request, read request body, make sure body length is not larger than declared, and send the response back to the client. It also provides some safeguards to ensure messages of the right type is sent in the right state. For example, it ensures that `http.response.start` messages are always sent before `http.response.body`, and no more messages are sent after connection closed. Most of the heavy lifting in parsing header, cookie, and query is done by `werkzeug` and other Python built-in superheroes. 

Let's make a simple ASGI application with the `Connection` class:
```python
# example1.py
from aaf.connection import Connection
async def app(scope, receive, send):
    conn = Connection(scope, receive=receive, send=send)
    name = conn.query.get("name")
    await conn.send("Hello, " + (name or "world"), finish=True)
```
We can run this example by executing `uvicorn example_1:app`. Requesting `/?name=foo` should correctly return `Hello, foo`. 

We now have an ASGI application that's shorter in code and does more (accessing HTTP query). That's one step closer to what we want!

## HTTP Responses
It's usually not required to have fine control over when to send what in an HTTP request-response cycle, returning a response that knows how to set headers and send the body is more convenient and familiar. To do that, we can write a simple `HttpResponse` helper class.

A `JsonResponse` is also added while we are at it.

```python
# response.py
import json
from typing import Union, Optional, Mapping, Any

from .connection import Connection


class HttpResponse:
    def __init__(
        self,
        body: Optional[Union[bytes, str]] = b"",
        connection: Optional[Connection] = None,
        *,
        status_code: int = 200,
        headers: Optional[Mapping[str, str]] = None
    ):
        self.body = body
        self.connection = connection
        self.status_code = status_code
        self.headers = headers

    def __await__(self):
        if not self.connection:
            raise ValueError("No connection")
        self.connection.resp_status_code = self.status_code
        if self.headers:
            for k, v in self.headers.items():
                self.connection.put_resp_header(k, v)
        return self.connection.send(self.body, finish=True).__await__()


class JsonResponse(HttpResponse):
    def __init__(
        self, data: Any, connection: Optional[Connection] = None, *args, **kwargs
    ):
        body = json.dumps(data)
        headers = kwargs.get("headers")
        if headers is None:
            headers = {}
        headers["content-type"] = "application/json"
        super().__init__(body, connection, *args, **kwargs)
```
There's not much going on in the `HttpResponse` class. All it does is providing a familiar interface allowing us to pass in a response body,  optional headers, optional status code, and calls the underneath methods in `Connection` class for us. In the example of `JsonResponse` class, it also sets the `content-type` header. 

Let's write another ASGI application to test it:
```python
async def app(scope, receive, send):
    conn = Connection(scope, receive=receive, send=send)
    await JsonResponse(conn.query.to_dict(flat=False), conn)
```
This application should return all query parameters in the form of a JSON object when visited. 

Great, the application is even shorter!

You might have noticed that this is not exactly how it's used in the [_Goal_ ](#goal) section, where we can just `return` the response object instead of `await` on it. This is because this example is a plain ASGI application and the one in the original [_Goal_ ](#goal) section is in the context of a `Router`, who calls the `await` for us. The `connection` argument is allowed to be `None` in the constructor for the same reason.

## Routing
Router dispatches requests based on requested url and HTTP method to different handlers. Most router implementations also parses parameters in urls. For example, if we define a router 
```python
@router.route('/a/<param_a>/<param_b>')
async def handler(connection, param_a, param_b):
	...
```
and then tell the router to match the URL `/a/foo/bar`, it should give us the `handler` function as well as the parameters `param_a` and `params_b`.

This is indeed not easy but luckily, `werkzeug` comes with a `routing` module with does exactly this and even more, such as automatically redirect in the case of missing trailing slash. With its help, we can implement our `routing` module in around 60 lines of code.
```python
# routing.py
import functools
from typing import Callable, Iterable, Optional

from werkzeug.routing import Map, MethodNotAllowed, NotFound, RequestRedirect, Rule

from .connection import Connection
from .response import HttpResponse


class Router:
    def __init__(self):
        super().__init__()
        self.url_map = Map()
        self.endpoint_to_handler = {}

    def route(self, rule, methods=None, name=None):
        methods = set(methods) if methods is not None else None
        if methods and not "OPTIONS" in methods:
            methods.add("OPTIONS")

        def decorator(name: Optional[str], handler: Callable):
            self.add_route(
                rule_string=rule, handler=handler, methods=methods, name=name
            )
            return handler

        return functools.partial(decorator, name)

    def add_route(
        self,
        *,
        rule_string: str,
        handler: Callable,
        name: Optional[str] = None,
        methods: Optional[Iterable[str]] = None,
    ):
        if not name:
            name = handler.__name__
        existing_handler = self.endpoint_to_handler.get(name)
        if existing_handler and existing_handler is not handler:
            raise ValueError("Duplicated route name: %s" % (name))
        self.url_map.add(Rule(rule_string, endpoint=name, methods=methods))
        self.endpoint_to_handler[name] = handler

    def get_url_binding_for_connection(self, connection: Connection):
        scope = connection.scope
        return self.url_map.bind(
            connection.req_headers.get("host"),
            path_info=scope.get("path"),
            script_name=scope.get("root_path") or None,
            url_scheme=scope.get("scheme"),
            query_args=scope.get("query_string", b""),
        )

    async def __call__(self, connection: Connection):
        try:
            rule, args = self.get_url_binding_for_connection(connection).match(
                return_rule=True, method=connection.scope.get("method")
            )
        except RequestRedirect as e:
            connection.resp_status_code = 302
            connection.put_resp_header("location", e.new_url)
            return await connection.send(f"redirecting to: {e.new_url}", finish=True)
        except MethodNotAllowed:
            connection.resp_status_code = 405
            return await connection.send(b"", finish=True)
        except NotFound:
            pass
        else:
            handler = self.endpoint_to_handler[rule.endpoint]
            res = await handler(connection, **args)
            if isinstance(res, HttpResponse):
                res.connection = connection
                await res

```
In the `__call__` method, I'm checking if the returned type from the handler is `HttpResponsoe`, if it is indeed `HttpResponsoe`, the router can `await` the response and the response takes care of sending headers and body.

> [More details on][werkzeug.routing] [werkzeug.routing][werkzeug.routing]

Use this simple app to test it:
```python
from aaf.connection import Connection
from aaf.response import JsonResponse
from aaf.routing import Router

router = Router()

@router.route("/hello/<name>")
async def hello(connection, name):
    return JsonResponse({'hello': name})


async def app(scope, receive, send):
    conn = Connection(scope, receive=receive, send=send)
    await router(conn)
```
Visiting `/hello/world` should return `{"hello": "world"}`. 

The application gets longer, but hopefully, it's clear that with more routes to handle a router makes it a lot easier to handle. 

## Router to ASGI application
Now, routers are great, but they are still not ASGI applications. Even though we can always write an ASGI application by hand, make a `Connection` out of it, and then call a router with the connection, it just doesn't feel like an ASGI framework but more like an ASGI toolbox. 

So, let's write a helper function that turns a list of routers into an ASGI application. The idea of receiving a list of routers instead of just one is to allow an unhandled path to failover to the next router.

```python
# __init__.py
from typing import List, Callable
from .connection import Connection


def aaf(routers: List[Callable]):
    async def asgi_app(scope, receive, send):
        conn = Connection(scope, send=send, receive=receive)
        for router in routers:
            await router(conn)
            if conn.finished:
                return
        if conn.started:
            await conn.finish()
        else:
            conn.resp_status_code = 404
            await conn.send("Not found", finish=True)

    return asgi_app
```
It also includes a default 404 response.

With the help of this new function, the previous example can be rewritten as:
```python
from aaf.connection import Connection
from aaf.response import JsonResponse
from aaf.routing import Router
from aaf import aaf

router = Router()

@router.route("/hello/<name>")
async def hello(connection, name):
    return JsonResponse({'hello': name})

app = aaf([router])
```

In fact, the example defined in the [goal](#goal) section should also work now. Give it a try!

# Next steps

An infinite number of things can be added to this framework. On the top of my head, there are
1. WebSocket support
2. Lifecycle hooks
3. More response types
4. More `Connection` methods
5. Built-in static file handling
6. URL reversing
7. Flask's blueprint

I will write some of the things in the near future. And I wish this post could give readers enough information on where to start implementing them on their own.


[asgi]: https://asgi.readthedocs.io/
[daphne]: https://github.com/django/daphne
[channels]: https://github.com/django/channels
[dispatch]: https://netflixtechblog.com/introducing-dispatch-da4b8a2a8072
[fastapi]: https://fastapi.tiangolo.com/
[fast-api-pr]: https://github.com/tiangolo/fastapi/pull/26
[uvicorn]: https://www.uvicorn.org/
[werkzeug.routing]: https://werkzeug.palletsprojects.com/en/1.0.x/routing/