FastAPI + Redis example¶

This example shows how to use Dependency Injector with FastAPI and Redis.

The source code is available on the Github.

Application structure¶

Application has next structure:

./
├──fastapiredis/
│├──__init__.py
│├──application.py
│├──containers.py
│├──redis.py
│├──services.py
│└──tests.py
├──docker-compose.yml
├──Dockerfile
└──requirements.txt

Redis¶

Module redis defines Redis connection pool initialization and shutdown. See fastapiredis/redis.py:

fromtypingimport AsyncIterator
fromredis.asyncioimport from_url, Redis
async definit_redis_pool(host: str, password: str) -> AsyncIterator[Redis]:
 session = from_url(f"redis://{host}", password=password, encoding="utf-8", decode_responses=True)
 yield session
 session.close()
 await session.wait_closed()

Service¶

Module services contains example service. Service has a dependency on Redis connection pool. It uses it for getting and setting a key asynchronously. Real life service will do something more meaningful. See fastapiredis/services.py:

"""Services module."""
fromredis.asyncioimport Redis
classService:
 def__init__(self, redis: Redis) -> None:
 self._redis = redis
 async defprocess(self) -> str:
 await self._redis.set("my-key", "value")
 return await self._redis.get("my-key")

Container¶

Declarative container wires example service with Redis connection pool. See fastapiredis/containers.py:

"""Containers module."""
fromdependency_injectorimport containers, providers
from.import redis, services
classContainer(containers.DeclarativeContainer):
 config = providers.Configuration()
 redis_pool = providers.Resource(
 redis.init_redis_pool,
 host=config.redis_host,
 password=config.redis_password,
 )
 service = providers.Factory(
 services.Service,
 redis=redis_pool,
 )

Application¶

Module application creates FastAPI app, setup endpoint, and init container.

Endpoint index has a dependency on example service. The dependency is injected using Wiring feature.

Listing of fastapiredis/application.py:

"""Application module."""
fromtypingimport Annotated
fromfastapiimport Depends, FastAPI
fromdependency_injector.wiringimport Provide, inject
from.containersimport Container
from.servicesimport Service
app = FastAPI()
@app.api_route("/")
@inject
async defindex(
 service: Annotated[Service, Depends(Provide[Container.service])]
) -> dict[str, str]:
 value = await service.process()
 return {"result": value}
container = Container()
container.config.redis_host.from_env("REDIS_HOST", "localhost")
container.config.redis_password.from_env("REDIS_PASSWORD", "password")
container.wire(modules=[__name__])

Tests¶

Tests use Provider overriding feature to replace example service with a mock. See fastapiredis/tests.py:

"""Tests module."""
fromunittestimport mock
importpytest
fromhttpximport ASGITransport, AsyncClient
from.applicationimport app, container
from.servicesimport Service
@pytest.fixture
defclient(event_loop):
 client = AsyncClient(
 transport=ASGITransport(app=app),
 base_url="http://test",
 )
 yield client
 event_loop.run_until_complete(client.aclose())
@pytest.mark.asyncio
async deftest_index(client):
 service_mock = mock.AsyncMock(spec=Service)
 service_mock.process.return_value = "Foo"
 with container.service.override(service_mock):
 response = await client.get("/")
 assert response.status_code == 200
 assert response.json() == {"result": "Foo"}

Sources¶