Why there is no App.state in Xpresso
Tags: python, webdev, flask, fastapi
Published: 2022-02-06 18:56:39

At some point, I'll write a longer blog post to introduce [Xpresso](https://github.com/adriangb/xpresso), but today I want to give only a brief background and then dive into a single design choice that I found interesting.

For a quick background, Xpresso is a Python web framework largely inspired by FastAPI, but with a more powerful dependency injection system and several other changes that you can read about [here](https://github.com/adriangb/xpresso#inspiration-and-relationship-to-other-frameworks).

Today I just want to talk about a single part of the framework though: `App.state`, or rather, the _lack_ of a `.state` like you can find in FastAPI (`fastapi.FastAPI.state`) or Starlette (`starlette.applications.Starlette.state`).

Let's start by taking a step back and look at why and how web frameworks store state.

## What is application state used for?

In most microframeworks (including Flask, Starlette FastAPI and Xpresso) the framework does not manage things like database connections for you.
Therefore, _you_ need to create a database connection and share it amongst requests.

There's several ways you could accomplish this:

- Use global variables (there are several issues with this)
- Use a framework provided global state (this is what Flask does)
- Use the `app` instance (this is what FastAPI and Starlette do)
- Use the dependency injection system (this is what Xpresso does)

## Case study: connection pool

When working with databases, especially PostgreSQL, you often use a connection pool. A connection pool lets you share a fixed number of connections to your database amongst all incoming requests, which makes acquiring a connection faster (thus reducing latency for the client) and lessens the load on your database.

But all you need to know here is that a connection pool is some sort of object which you need to access in every request, needs to be persisted in memory across requests and often can't be shared across processes / event loops.

I am choosing connection pools as an example because it is a non-trivial but common use case that has a reasonable amount of complexity attached to it (thread safety, a complex object not just some dictionary, etc.).

But we won't be using any specific database driver, instead let's imagine you have a `db.py` file with something like this:

```python
class Connection:
    def execute(self, query: str) -> None:
        ...

class Pool:
    def acquire(self) -> Connection:
        ...
```

For simplicity, we'll ignore the whole sync / async thing, it doesn't make much of a difference, although if you were using an ASGI frameworks (FastAPI, Starlette or Xpresso) you'd probably pick an async driver (the API above is sync, which would work best for WSGI framework like Flask).

Now we can look at what it takes to create and persist this `Pool` object across connections in several different frameworks.

## Flask

```python
from flask import Flask, current_app, g
from db import Pool

home = Blueprint('home', __name__, url_prefix='/')

@home.route("/")
def hello_world():
    conn = get_connection()
    conn.execute("SELECT * FROM table")
    return "<p>Hello, World!</p>"

def create_app():
    app = Flask(__name__)
    app.pool = Pool()
    app.register_blueprint(home)

def get_connection():
    if not hasattr(g, "connection"):
        g.connection = current_app.pool.acquire()
    return g.connection
```

I admit I am not a Flask expert, so please leave a comment if you think there's a better way to do this.

Flask gives you a `current_app` object, which points to the `Flask` object that is fulfilling the current request, and a `g` object, which is global to that particular request.

But there is no tracking of what is in these objects, and it certainly is not typed, so you get no autocomplete for `Connection.execute` and no static analysis for error checking.

## FastAPI / Starlette

Starlette and FastAPI let you store arbitrary data in `app.state`, which you can access from a request via `Request.app.state`.

What you get back is a [State](https://github.com/encode/starlette/blob/424351cb231c67798a65c091b0b7d42790f5e444/starlette/datastructures.py#L658-L683) object, which is just a blank slate object that allows attribute assignment and lookup.

```python
from fastapi import Depends, FastAPI, Request
from db import Pool, Connection

app = FastAPI()

@app.on_startup
async def app_startup():
   app.state.pool = Pool()

def get_connection(request: Request) -> Connection:
    return request.app.state.pool.acquire()

@app.get("/")
async def home(conn: Connection = Depends(get_connection)):
    conn.execute("SELECT * FROM table")
    return "Hello world!"
```

This works, but we have some boilerplate from the startup event, and `request.app.state.pool` is not typed.

## Xpresso

In Xpresso, you get to define your own app state:

```python
from contextlib import asynccontextmanager
from dataclasses import dataclass

from xpresso import App, Path, Depends
from xpresso.typing import Annotated

from db import Pool, Connection

@dataclass
class AppState:
    pool: Pool | None = None

@asynccontextmanager
async def lifespan(state: AppState) -> None:
    state.pool = Pool() 

def get_connection(state: AppState) -> Connection:
    assert state.pool is not None
    return state.pool.acquire()

ConnDepends = Annotated[Connection, Depends(get_connection)]

async def home(conn: ConnDepends) -> str:
    conn.execute("SELECT * FROM table")
    return "Hello world!"

app = App(lifespan=lifespan, routes=[Path("/", get=home)])
```

On the one hand everything is fully typed now.
On the other hand, it's quite a bit more verbose: we had to `assert state.pool` is not None and we had to add a lifespan event, which is a context manager.

But we can do better: we don't even _need_ and `AppState` object, Xpresso can manage this state for us implicitly.

```python
from typing import Annotated

from xpresso import App, Depends

from db import Pool, Connection

PoolDepends = Annotated[Pool, Depends(scope="app")]

def get_connection(pool: PoolDepends) -> Connection:
    return pool.acquire()

ConnDepends = Annotated[Connection, Depends(get_connection)]

async def home(conn: ConnDepends) -> str:
    conn.execute("SELECT * FROM table")
    return "Hello world!"

app = App(routes=[Path("/", get=home)])
```

All we had to do was tell Xpresso to tie the `Pool` object to the application's lifespan (`scope="app"`), which is a single line of code and preserves type safety.

And this is why there is no `App.state` or `current_app` in Xpresso: you simply don't need it.