Forem: Adrian Garcia Badaracco

Why there is no App.state in Xpresso

Adrian Garcia Badaracco — Sun, 06 Feb 2022 18:56:39 +0000

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.

di: pythonic dependency injection

Adrian Garcia Badaracco — Fri, 01 Oct 2021 16:12:42 +0000

In this article, I would like to introduce di, a dependency injection framework that seeks to be:

Intuitive: simple things are easy, complex things are possible
Succinct: low boilerplate, can run your code directly via autowiring
Correct: tested with MyPy; exports valid type annotations
Powerful: with lifecycles, shared dependencies and a flexible API
Performant: supporting async dependencies and concurrent execution

Introduction to dependency injection

Dependency injection is a technique for organizing larger applications by handing over control of your dependencies to "something" (often a framework) that assembles them for you.
It's a complicated topic, so maybe it's better to just look at an example.
We'll build our example around a pretty common scenario: you have a class that needs to make web requests, so it needs an HTTP client (we'll use httpx).

from httpx import Client


class WeatherClient:
    def __init__(self) -> None:
        self.client = Client()
    def get_temp(self, lat: float, long: float) -> float:
        self.client.get(...)
        ...

This class does not use dependency injection.
The problem arises when we want to test this class: we probably don't want to make real web requests in our unit test.

Luckily, httpx offers the ability to use mock transports:

import httpx

def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"temp": 43.21})

client = httpx.Client(transport=httpx.MockTransport(handler))

But we have a problem: how do we get our client using a mock transport into our WeatherClient?
One approach often used is monkey patching.
While monkey patching has its use cases, it can also lead to a lot of very confusing problems and messy code, especially in a larger project.

Dependency injection gives us an alternative: have WeatherClient accept an instance of httpx.Client:

from typing import Optional

from httpx import Client


class WeatherClient:
    def __init__(self, client: Optional[Client] = None) -> None:
        self.client = client or Client()

This form of dependency injection is called constructor injection because we inject the dependency in the constructor / __init__ method.

In this case, we chose to add a None default so that we can detect when the caller did not include a Client and so we can make a default one.
This is largely a matter of preference, and depends on the project you are working on.
For a library, this is probably a good idea.
For an application, maybe not so much.

Note: did you notice that httpx.Client itself uses dependency injection for the transport parameter? Neat huh!

The problems

As per above, dependency injection is not as crazy as it sounds, it can be simple.
And you're probably already using it in libraries like httpx.

But what are the downsides?

Well the main issue is that when you end up with deeply nested / a lot of dependencies, as would tend to happen in a larger project, you can end up with some really ugly constructor calls, or you may need the same dependency in multiple places, which means you need a temporary variable, etc.

Basically, the code to assemble and keep track of all of these dependencies can become a headache in and of itself.

Enter: dependency injection frameworks.

Dependency injection frameworks seek to solve this problem by abstracting away the boilerplate of creating and connecting dependencies together.

Dependency injection frameworks

Although dependency injection frameworks come in many shapes and sizes, in principle they all provide similar functionality:

Register dependencies and providers for those dependencies (i.e. tell the framework how to build your dependencies)
Inject your dependencies for you

Some frameworks also have more advanced features such as:

Autowiring: transitive dependencies are discovered from parameter names or type annotations
Lifecycles: the ability to hook into creation, deletion and error events during the lifecycle of your dependency.
Scopes: singleton dependencies or dependencies that are re-used within each call if they are needed in multiple places.

di is one of the latter frameworks, offering all of these features and more.

Using `di`

Let's start with a simple example.
We want to build a web app that uses our WeatherClient.

import os
from dataclasses import dataclass, field

from xyz import App


def url_from_env() -> str:
    return os.environ["WEATHER_URL"]


@dataclass
class AppConfig:
     weather_url: str = field(default_factory=url_from_env)


class MyApp:
    def __init__(self, weather_client: WeatherClient) -> None:
    self.weather_client = weather_client

Building this by hand might look something like:

app = App(
    weather_client=WeatherClient(
        client=httpx.Client(
            ...
        )
    ),
    config=AppConfig(
        ...
    )
)

Using di:

from di import Container, Dependant


container = Container()
app = container.execute_sync(
    container.solve(Dependant(App))
)

Notice that it doesn't matter how deep the dependencies go or how many there are, we only ever said "we want an App instance".

What about the injection of the Client?

client = Client(
    transport=MockTransport(handler=handler)
)
container.bind(Dependant(lambda: client), Client)
app = container.execute_sync(
    container.solve(Dependant(App))
)

Again, the important thing here is that Client could be nested 10 layers of dependencies deep and the above code would not change at all.

Advanced functionality

Aside from the basic usage above, di support a lot more functionality.

Here are some snippets demonstrating the functionality, but for full fledged examples and more explanation, see our docs.

Dependency sharing

from di import Container, Depends, Dependant

def func() -> object:
    return object()

def dependant(
    one: object = Depends(func),
    two: object = Depends(func),
    three: object = Depends(func, share=False)
) -> None:
    assert one is two and two is not three

container = Container()
container.execute_sync(container.solve(Dependant(dependant)))

In this example, func is called twice (by default in two separate threads), once for one and two, which share the same value, and once for three.

Scopes and lifecycles

from typing import Generator

from di import Container, Depends, Dependant

def func() -> Generator[int, None, None]:
   print("func startup")
   yield 1
   print("func shutdown")


def dependant(v: int = Depends(func, scope=1234)) -> int:
    print("computing")
    return v + 1

container = Container()
with container.enter_local_scope(1234):
    print("enter scope")
    res = container.execute_sync(
       container.solve(Dependant(dependant))
    )
    print("exit scope")

assert res == 2

This example will print:

enter scope
func startup
computing
func shutdown
exit scope

And is equivalent to:

from contextlib import contextmanager


with contextmanager(func)() as value:
    dependant(v=value)

Scopes in di are arbitrary: any hashable value will do.
As long as you are consistent in declaring scopes, di will not care what you use. If "you" are a framework calling into user code, this gives you the freedom to define and manage your own scopes. For a web framework, these might be "app" and "request". For a CLI, there might be a single "call" scope. For a TUI, there could be an "app" and "event scope"

Solved dependencies

Often, you have to run the same dependency multiple times but aren't dynamically changing the dependency graph every execution (you would be dynamically changing the graph if you were doing container.bind between each call).

In these scenarios, di lets you save the entire computed dependency graph into a SavedDependency and execute the same graph multiple times.
In fact, this is what we've been doing already every time we called container.solve.

# solving locks in binds, but not cached values or scopes
solved = container.solve(Dependant(dependency))
res1 = container.execute_sync(solved)
res2 = container.execute_sync(solved)

Async execution

Simply swap out execute_sync for execute_async for async support:

from di import Container, Dependant


async def dep() -> int:
    return 1

async def main() -> None:
    container = Container()
    solved = container.solve(Dependant(dep))
    res = await container.execute_async(solved)
    assert res == 1

Like sync execution, async execution supports arbitrary callables as well as generator functions that get used as context managers.

DependantProtocol

The DependantProtocol interface provides an abstract representation of a dependency specification.
The Container uses this abstract representation to build dependency graphs.

This means that you can easily customize a lot of the behavior in di by overriding this API.
If you want to use some of the built in functionality and just tweak it slightly, you can inherit from di.Dependant, but you don't have to.

For example, to accept a default value instead of a callable:

from di import Dependant
from di.types.providers import DependencyType
from di.types.scopes import Scope

class DefaultDependant(Dependant[DependencyType]):
    def __init__(
        self,
        default: DependencyType,
        scope: Scope,
        shared: bool,
    ) -> None:
        super().__init__(
            call=lambda: default,
            scope=scope,
            share=share,
         )

Now if there is no bind / override on the dependency, it will execute just like if it were being called directly.

Executors

One of the design principles of di is to isolate the work of the container into distinct steps:

Wiring, which can be controlled via the DependantProtocol
Solving, which is owned by the container
Execution, which is managed via the Executor API

To this end, di allows you to provide your own executor, which only needs to take a list of groups of tasks (callable functions that take no arguments) and execute them in the specified order.

The Executor API is as follows:

import typing

ResultType = typing.TypeVar("ResultType")

Task = typing.Union[
    typing.Callable[[], None],
    typing.Callable[[], typing.Awaitable[None]]
]


class SyncExecutor(typing.Protocol):
    def execute_sync(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        ...


class AsyncExecutor(typing.Protocol):
    async def execute_async(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        ...

The default executor (di.executors.DefaultExecutor) supports both sync and async execution.
Dependencies are executed concurrently (via threads in sync execution and tasks in async execution) for dependencies that can be executed in concurrently.

A simple SyncExecutor implementation might look like:

class SimpleExecutor(SyncExecutor):
    def execute_sync(
        self,
        tasks: typing.List[typing.List[Task]],
        get_result: typing.Callable[[], ResultType],
    ) -> ResultType:
        for group in tasks:
            for task in group:
                res = task()
                if inspect.isawaitable(res):
                    raise TypeError
         return get_result()

Note that SyncExecutor may be handed async functions as tasks.
The sync/async portion refers to the API of the executor itself, not the the tasks it may be handed.
It is up to the implementation to raise an exception or execute the async task somehow.

Integration into other libraries

While di should be simple enough that it can be used by end users, it shines when integrated into a framework or library that fully embraces inversion of control.

To this end, we provide sample integrations, currently only for Starlette and Textual, but more ideas are welcome!

Performance

We provide some basic benchmarks comparing to FastAPI.

The results strongly depend on the size of the DAG.

For a relatively large but not completely absurd DAG of 12 vertices with each vertex having 0-3 edges, di performs slightly faster than FastAPI (a couple percent per request).

For larger DAGs, especially where di is able to execute dependencies concurrently, di can be 10x faster than FastAPI.
To be fair, this is not a real world scenario, and FastAPI does a lot more than the Starlette di integration is doing (keeping track of stuff for OpenAPI, etc.).
If these benchmarks are unfair for other reasons, please let me know.

Conclusion

Dependency injection and inversion of control are powerful techniques, but it can get complicated.
Python has shown time and time again that it can take complicated things and make them simple for users.
FastAPI is one of the best examples of this.

I hope that di can help other libraries, frameworks or projects use inversion of control and dependency injection without having to go through the pain of writing their own dependency injection system.