FastAPI example

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

    config = providers.Configuration()

    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 Optional, List

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from dependency_injector.wiring import inject, Provide

from .services import SearchService
from .containers import Container


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(
        query: Optional[str] = None,
        limit: Optional[str] = None,
        default_query: str = Depends(Provide[Container.config.default.query]),
        default_limit: int = Depends(Provide[Container.config.default.limit.as_int()]),
        search_service: SearchService = Depends(Provide[Container.search_service]),
):
    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, creates FastAPI 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.from_yaml('config.yml')
    container.config.giphy.api_key.from_env('GIPHY_API_KEY')
    container.wire(modules=[endpoints])

    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
from httpx import AsyncClient

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


@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):
    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.