Sanic example¶
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, creates
Sanic 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
@pytest.fixture
def app():
Sanic.test_mode = True
app = create_app()
yield app
app.ctx.container.unwire()
@pytest.fixture
def test_client(loop, app, sanic_client):
return loop.run_until_complete(sanic_client(app))
async def test_index(app, test_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.ctx.container.giphy_client.override(giphy_client_mock):
response = await test_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, test_client):
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 test_client.get("/")
assert response.status_code == 200
data = response.json()
assert data["gifs"] == []
async def test_index_default_params(app, test_client):
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 test_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()
