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 (
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
appinstance (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: ...
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.
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
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!"
This works, but we have some boilerplate from the startup event, and
request.app.state.pool is not typed.
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)])
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)])
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
current_app in Xpresso: you simply don't need it.
Top comments (0)