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:

from typing import AsyncIterator from redis.asyncio import from_url, Redis async def init_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.""" from redis.asyncio import Redis class Service: def __init__(self, redis: Redis) -> None: self._redis = redis async def process(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.""" from dependency_injector import containers, providers from . import redis, services class Container(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.""" from typing import Annotated from fastapi import Depends, FastAPI from dependency_injector.wiring import Provide, inject from .containers import Container from .services import Service app = FastAPI() @app.api_route("/") @inject async def index( 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.""" from unittest import mock import pytest from httpx import ASGITransport, AsyncClient from .application import app, container from .services import Service @pytest.fixture def client(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 def test_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¶