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

This example shows how to use Dependency Injector with Sanic.

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 │ ├── main.py │ ├── application.py │ ├── containers.py │ ├── giphy.py │ ├── handlers.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=[".handlers"])

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

Handlers¶

Handler has dependencies on search service and some config options. The dependencies are injected using Wiring feature.

Listing of giphynavigator/handlers.py:

"""Handlers module."""

from sanic.request import Request from sanic.response import HTTPResponse, json from dependency_injector.wiring import inject, Provide

from .services import SearchService from .containers import Container

@inject async def index( request: Request, search_service: SearchService = Provide[Container.search_service], default_query: str = Provide[Container.config.default.query], default_limit: int = Provide[Container.config.default.limit.as_int()], ) -> HTTPResponse: query = request.args.get("query", default_query) limit = int(request.args.get("limit", default_limit))

gifs = await search_service.search(query, limit)

return json(
    {
        "query": query,
        "limit": limit,
        "gifs": gifs,
    },
)

Application factory¶

Application factory creates container, wires it with the handlers module, createsSanic app and setup routes.

Listing of giphynavigator/application.py:

"""Application module."""

from sanic import Sanic

from .containers import Container from . import handlers

def create_app() -> Sanic: """Create and return Sanic application.""" container = Container() container.config.giphy.api_key.from_env("GIPHY_API_KEY")

app = Sanic("giphy-navigator")
app.ctx.container = container
app.add_route(handlers.index, "/")
return 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 sanic import Sanic

from giphynavigator.application import create_app from giphynavigator.giphy import GiphyClient

pytestmark = pytest.mark.asyncio

@pytest.fixture def app(): Sanic.test_mode = True app = create_app() yield app app.ctx.container.unwire()

async def test_index(app): 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.ctx.container.giphy_client.override(giphy_client_mock):
    _, response = await app.asgi_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"},
    ],
}

async def test_index_no_data(app): giphy_client_mock = mock.AsyncMock(spec=GiphyClient) giphy_client_mock.search.return_value = { "data": [], }

with app.ctx.container.giphy_client.override(giphy_client_mock):
    _, response = await app.asgi_client.get("/")

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

async def test_index_default_params(app): giphy_client_mock = mock.AsyncMock(spec=GiphyClient) giphy_client_mock.search.return_value = { "data": [], }

with app.ctx.container.giphy_client.override(giphy_client_mock):
    _, response = await app.asgi_client.get("/")

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

Sources¶

Explore the sources on the Github.