DEV Community

Adrian Garcia Badaracco
Adrian Garcia Badaracco

Posted on

Why there is no App.state in Xpresso

At some point, I'll write a longer blog post to introduce 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.

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:

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

class Pool:
    def acquire(self) -> Connection:
        ...
Enter fullscreen mode Exit fullscreen mode

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

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
Enter fullscreen mode Exit fullscreen mode

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 object, which is just a blank slate object that allows attribute assignment and lookup.

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!"
Enter fullscreen mode Exit fullscreen mode

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:

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)])
Enter fullscreen mode Exit fullscreen mode

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.

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)])
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)