FastAPI + Redis example — Dependency Injector 4.46.0 documentation (original) (raw)

This example shows how to use Dependency Injector with FastAPI andRedis.

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¶

The source code is available on the Github.