FastAPI + Redis example

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

The source code is available on the Github.

See also:

Application structure

Application has next structure:

├── fastapiredis/
│   ├──
│   ├──
│   ├──
│   ├──
│   ├──
│   └──
├── docker-compose.yml
├── Dockerfile
└── requirements.txt


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

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
    await session.wait_closed()


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 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")


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

"""Containers module."""

from dependency_injector import containers, providers

from . import redis, services

class Container(containers.DeclarativeContainer):

    config = providers.Configuration()

    redis_pool = providers.Resource(

    service = providers.Factory(


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 module."""

from dependency_injector.wiring import inject, Provide
from fastapi import FastAPI, Depends

from .containers import Container
from .services import Service

app = FastAPI()

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")


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

"""Tests module."""

from unittest import mock

import pytest
from httpx import AsyncClient

from .application import app, container
from .services import Service

def client(event_loop):
    client = AsyncClient(app=app, base_url="http://test")
    yield client

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"}


The source code is available on the Github.

See also: