How to use FastAPI with an IoC framework
While FastAPI has a dependency injection mechanism, it doesn't implement a comprehensive IoC. The aim of this article is to utilize an IoC framework without leaking abstractions between layers of the software.
Problem Definition
Consider we have a Protocol and its implementation and we want to inject the concrete class into our API handler method.
class Greeter(Protocol):
def greet(self) -> str: ...
class WorldGreeter(Greeter):
def greet(self) -> str:
return 'Hello World!'
There are IoC frameworks for Python that we can register this Protocol and its corresponding concrete into them. Suppose using punq for example:
def get_container() -> punq.Container:
container = punq.Container()
container.register(Greeter, WorldGreeter)
return container
container = get_container()
The question is how can we use this container with FastAPI.
Step 1: Using a higher-order function
FastAPI has a dependency injection mechanism. This mechanism accepts a function as an argument. Let’s consider this example:
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(paging_parameters)]):
...
Before handling each request, Depends
tries invoking paging_parameters
and passing the result of it as the commons
parameter to the request handler (read_items
here). Notice that paging_parameters
was passed without parentheses as a function parameter.
How can we take advantage of this on our end? We want to pass the resolve
method of the container
object. But that method takes the class we want to resolve as a parameter. So we want to have a parameterless function as the parameter of Depends
and that function should somehow know what was the parameter for resolve
. Sounds like closure? Because it is:
T = TypeVar('T')
def resolver(t: Type[T]) -> Callable[[], T]:
return lambda: container.resolve(t)
app = FastAPI()
@app.get("/greet")
async def greet(greeter: Annotated[Greeter, Depends(resolver(Greeter))]):
return {"message": greeter.greet()}
Here we created a resolver function whose return value is a function. When we defined our greet
handler, we passed resolver(Greeter)
which returns the lambda function that uses container
to resolve the Greeter
class.
It is worth noting that we could equally define our resolver
function like this:
def resolver(t: Type[T]) -> Callable[[], T]:
return functools.partial(container.resolve, t)
Step 2: It is a Service
Although the previous method works fine, typing Annotated[Greeter, Depends(resolver(Greeter))]
each time might be error-prone. What if we had a function that could be replaced with Depends(resolver(Greeter))
?
def Service(t: Type[T]) -> Any: # noqa: N802
def resolver(t: Type[T]) -> Callable[[], T]:
return lambda: container.resolve(t)
return Depends(resolver(t))
Using this Service
function, we can rewrite our handler simply as:
@app.get("/greet")
async def greet(greeter: Annotated[Greeter, Service(Greeter)]):
return {"message": greeter.greet()}
Neat and concise.
Step 3: Larger Programs with multiple Files
In larger programs, we usually define our endpoints in multiple files. On the other hand, the way we defined our Service
function, it directly depends on our IoC container and because of this, it cannot be imported into other files.
To remedy this issue, first, we add our container to the state of our FastApi app:
app.state.ioc_container = container
Second, we update our resolver
function to take a second argument so that we can access our container through it:
def Service(t: Type[T]) -> Any: # noqa: N802
def resolver(t: Type[T], request: Request) -> T:
return request.app.state.ioc_container.resolve(t)
return Depends(functools.partial(resolver, t))
Note that functools.partial(resolver, t)
is a single argument function that takes a Request
parameter. But the point is that FastAPI knows how to inject Request
parameters.
Conclusion
We created a small handy function that can be used anywhere to inject services into our handler functions.
You can find the entire program here or in this gist.
import functools
from typing import Annotated, Any, Callable, Protocol, Type, TypeVar
from fastapi import APIRouter, Depends, FastAPI, Request
import punq
# domain - no dependency to FastAPI or IoC
class Greeter(Protocol):
def greet(self) -> str: ...
class WorldGreeter(Greeter):
def greet(self) -> str:
return 'Hello World!'
# IoC
def get_container() -> punq.Container:
container = punq.Container()
container.register(Greeter, WorldGreeter)
return container
# Helper - No dependency to app or container
T = TypeVar('T')
def Service(t: Type[T]) -> Any: # noqa: N802
def resolver(t: Type[T], request: Request) -> Callable[[], T]:
return request.app.state.ioc_container.resolve(t)
return Depends(functools.partial(resolver, t))
# API - No dependency to app or container
router = APIRouter()
@router.get("/greet")
async def greet(greeter: Annotated[Greeter, Service(Greeter)]):
return {"message": greeter.greet()}
# app
app = FastAPI()
app.state.ioc_container = get_container()
app.include_router(router)