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/
│   ├── __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:

"""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
    pool.close()
    await pool.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', encoding='utf-8')

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

import sys

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

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=[sys.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 AsyncClient

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


@pytest.fixture
def client(event_loop):
    client = AsyncClient(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

The source code is available on the Github.

See also: