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 aioredis 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 aioredis 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 dependency_injector.wiring import inject, Provide
from fastapi import FastAPI, Depends

from .containers import Container
from .services import Service


app = FastAPI()


@app.api_route("/")
@inject
async def index(service: Service = Depends(Provide[Container.service])):
    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¶