FastAPI Integration

FastAPI has a great dependency system, but one of the key features it lacks is the ability to create application or singleton scoped dependencies which are not bound to the globals of a module. With flexdi you can create application scoped dependencies using the FastAPIGraph class.

FastAPIGraph is an extension of the normal FlexGraph with the nessisary logic to hook itself into FastAPI’s startup and shutdown. Upon getting a request, the FlexGraph will begin a new request scope to allow creating dependencies that have the same lifetime rules as the request, but it can inherit any application scoped dependencies which are created at server startup time.

To use a FlexGraph proided dependency within a FastAPI dependency or route, you can define the default value of your argument as a FlexDepends instead of the normal fastapi.Depends instance. The argument to FlexDepends is the class which you want to be injected as the value.

from typing import AsyncIterator

from fastapi import FastAPI
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine

from flexdi.fastapi import FastAPIGraph, FlexDepends

# Creating the graph with reference to the FastAPI app binds the graph to the
# startup and shutdown events of the server, and enables FlexDepends values to
# be resolvable from the request context.
app = FastAPI()
graph = FastAPIGraph(app)


# Setting this dependency as application scopes will mean that it gets created
# once during the lifetime of the server, and will be re-used for every request.
# Anything that is application scoped MUST be thread-safe if you are going to use
# it from non asyncio endpoints.
@graph.bind(scope="application", eager=True)
async def provide_engine() -> AsyncIterator[AsyncEngine]:
    engine = create_async_engine("sqlite+aiosqlite:///mydb.db")
    try:
        yield engine
    finally:
        await engine.dispose()


# Here the connection dependency is not set to application scoped.
# It will be created uniquely for each request we get.
@graph.bind
async def provide_connection(engine: AsyncEngine) -> AsyncIterator[AsyncConnection]:
    async with engine.begin() as conn:
        yield conn


# FlexDepends will make fastapi defer to the FlexGraph for creating the instance.
@app.get("/")
async def get(conn: AsyncConnection = FlexDepends(AsyncConnection)) -> list[str]:
    statement = text("SELECT name FROM sqlite_master;")
    return [table_name for [table_name] in await conn.execute(statement)]


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=5007)

Note

Sub-dependencies of flexdi injected arguments do not have access to the request context or other FastAPI specific dependencies, such as bodies, query params, headers, etc. They only have access to dependencies registed in the FlexGraph bindings.

Warning

Flex dependencies will all be created on the main server thread, meaning dependencies with expensive startup costs will slow down your server.