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