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."""
fromdependency_injectorimport containers, providers
from.import giphy, services
classContainer(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."""
fromtypingimport Annotated, List
fromfastapiimport APIRouter, Depends
frompydanticimport BaseModel
fromdependency_injector.wiringimport Provide, inject
from.containersimport Container
from.servicesimport SearchService
classGif(BaseModel):
 url: str
classResponse(BaseModel):
 query: str
 limit: int
 gifs: List[Gif]
router = APIRouter()
@router.get("/", response_model=Response)
@inject
async defindex(
 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, creates FastAPI app, and setup routes.

Listing of giphynavigator/application.py:

"""Application module."""
fromfastapiimport FastAPI
from.containersimport Container
from.import endpoints
defcreate_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."""
fromunittestimport mock
importpytest
importpytest_asyncio
fromhttpximport ASGITransport, AsyncClient
fromgiphynavigator.applicationimport app
fromgiphynavigator.giphyimport GiphyClient
@pytest_asyncio.fixture
async defclient():
 async with AsyncClient(
 transport=ASGITransport(app=app),
 base_url="http://test",
 ) as client:
 yield client
@pytest.mark.asyncio
async deftest_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 deftest_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 deftest_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.

Sponsor the project on GitHub:

[フレーム]