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:
...
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
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!"
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)])
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 App.state
or current_app
in Xpresso: you simply don't need it.
Top comments (0)