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/

"""Redis client module."""

from typing import AsyncIterator

from aioredis import create_redis_pool, Redis

async def init_redis_pool(host: str, password: str) -> AsyncIterator[Redis]:
    pool = await create_redis_pool(f'redis://{host}', password=password)
    yield pool
    await pool.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', encoding='utf-8')


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

import sys

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

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: