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

This example shows how to use Dependency Injector with FastAPI.

The example application is a REST API that searches for funny GIFs on the Giphy.

The source code is available on the Github.

Application structure¶

Application has next structure:

./ ├── giphynavigator/ │ ├── init.py │ ├── application.py │ ├── containers.py │ ├── endpoints.py │ ├── giphy.py │ ├── services.py │ └── tests.py ├── config.yml └── requirements.txt

Container¶

Declarative container is defined in giphynavigator/containers.py:

"""Containers module."""

from dependency_injector import containers, providers

from . import giphy, services

class Container(containers.DeclarativeContainer):

wiring_config = containers.WiringConfiguration(modules=[".endpoints"])

config = providers.Configuration(yaml_files=["config.yml"])

giphy_client = providers.Factory(
    giphy.GiphyClient,
    api_key=config.giphy.api_key,
    timeout=config.giphy.request_timeout,
)

search_service = providers.Factory(
    services.SearchService,
    giphy_client=giphy_client,
)

Endpoints¶

Endpoint has a dependency on search service. There are also some config options that are used as default values. The dependencies are injected using Wiring feature.

Listing of giphynavigator/endpoints.py:

"""Endpoints module."""

from typing import Annotated, List

from fastapi import APIRouter, Depends from pydantic import BaseModel

from dependency_injector.wiring import Provide, inject

from .containers import Container from .services import SearchService

class Gif(BaseModel): url: str

class Response(BaseModel): query: str limit: int gifs: List[Gif]

router = APIRouter()

@router.get("/", response_model=Response) @inject async def index( default_query: Annotated[str, Depends(Provide[Container.config.default.query])], default_limit: Annotated[ int, Depends(Provide[Container.config.default.limit.as_int()]) ], search_service: Annotated[ SearchService, Depends(Provide[Container.search_service]) ], query: str | None = None, limit: int | None = None, ): query = query or default_query limit = limit or default_limit

gifs = await search_service.search(query, limit)

return {
    "query": query,
    "limit": limit,
    "gifs": gifs,
}

Application factory¶

Application factory creates container, wires it with the endpoints module, createsFastAPI app, and setup routes.

Listing of giphynavigator/application.py:

"""Application module."""

from fastapi import FastAPI

from .containers import Container from . import endpoints

def create_app() -> FastAPI: container = Container() container.config.giphy.api_key.from_env("GIPHY_API_KEY")

app = FastAPI()
app.container = container
app.include_router(endpoints.router)
return app

app = create_app()

Tests¶

Tests use Provider overriding feature to replace giphy client with a mock giphynavigator/tests.py:

"""Tests module."""

from unittest import mock

import pytest import pytest_asyncio from httpx import ASGITransport, AsyncClient

from giphynavigator.application import app from giphynavigator.giphy import GiphyClient

@pytest_asyncio.fixture async def client(): async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: yield client

@pytest.mark.asyncio async def test_index(client): giphy_client_mock = mock.AsyncMock(spec=GiphyClient) giphy_client_mock.search.return_value = { "data": [ {"url": "https://giphy.com/gif1.gif"}, {"url": "https://giphy.com/gif2.gif"}, ], }

with app.container.giphy_client.override(giphy_client_mock):
    response = await client.get(
        "/",
        params={
            "query": "test",
            "limit": 10,
        },
    )

assert response.status_code == 200
data = response.json()
assert data == {
    "query": "test",
    "limit": 10,
    "gifs": [
        {"url": "https://giphy.com/gif1.gif"},
        {"url": "https://giphy.com/gif2.gif"},
    ],
}

@pytest.mark.asyncio async def test_index_no_data(client): giphy_client_mock = mock.AsyncMock(spec=GiphyClient) giphy_client_mock.search.return_value = { "data": [], }

with app.container.giphy_client.override(giphy_client_mock):
    response = await client.get("/")

assert response.status_code == 200
data = response.json()
assert data["gifs"] == []

@pytest.mark.asyncio async def test_index_default_params(client): giphy_client_mock = mock.AsyncMock(spec=GiphyClient) giphy_client_mock.search.return_value = { "data": [], }

with app.container.giphy_client.override(giphy_client_mock):
    response = await client.get("/")

assert response.status_code == 200
data = response.json()
assert data["query"] == app.container.config.default.query()
assert data["limit"] == app.container.config.default.limit()

Sources¶

Explore the sources on the Github.