Clever Geek Handbook

Aiohttp + Dependency Injector - Tutoriel d'injection de dépendances

Salut,



je suis le créateur de Dependency Injector . Il s'agit d'un framework d'injection de dépendances pour Python.



Poursuite d'une série de didacticiels sur l'utilisation de l'injecteur de dépendances pour créer des applications.



Dans ce didacticiel, je souhaite montrer comment utiliser l'injecteur de dépendances pour le développement d' aiohttpapplications.



Le manuel comprend les parties suivantes:



  1. Qu'allons-nous construire?
  2. Préparer l'environnement
  3. Structure du projet
  4. Installer les dépendances
  5. Application minimale
  6. Client API Giphy
  7. Service de recherche
  8. Connecter la recherche
  9. Un peu de refactoring
  10. Ajouter des tests
  11. Conclusion


Le projet terminĂ© peut ĂȘtre trouvĂ© sur Github .



Pour commencer, vous devez avoir:



  • Python 3.5+
  • Environnement virtuel


Et il est souhaitable d'avoir:



  • CompĂ©tences de dĂ©veloppement initial avec aiohttp
  • Comprendre le principe de l'injection de dĂ©pendances


Qu'allons-nous construire?







Nous allons créer une application API REST qui recherche des gifs amusants sur Giphy . Appelons cela Giphy Navigator.



Comment fonctionne Giphy Navigator?



  • Le client envoie une demande indiquant ce qu'il faut rechercher et le nombre de rĂ©sultats Ă  renvoyer.
  • Giphy Navigator renvoie la rĂ©ponse json.
  • La rĂ©ponse comprend:

    • requĂȘte de recherche
    • nombre de rĂ©sultats
    • Liste d'URL GIF


Exemple de réponse: 



{
    "query": "Dependency Injector",
    "limit": 10,
    "gifs": [
        {
            "url": "https://giphy.com/gifs/boxes-dependent-swbf2-6Eo7KzABxgJMY"
        },
        {
            "url": "https://giphy.com/gifs/depends-J56qCcOhk6hKE"
        },
        {
            "url": "https://giphy.com/gifs/web-series-ccstudios-bro-dependent-1lhU8KAVwmVVu"
        },
        {
            "url": "https://giphy.com/gifs/TheBoysTV-friends-friend-weneedeachother-XxR9qcIwcf5Jq404Sx"
        },
        {
            "url": "https://giphy.com/gifs/netflix-a-series-of-unfortunate-events-asoue-9rgeQXbwoK53pcxn7f"
        },
        {
            "url": "https://giphy.com/gifs/black-and-white-sad-skins-Hs4YzLs2zJuLu"
        },
        {
            "url": "https://giphy.com/gifs/always-there-for-you-i-am-here-PlayjhCco9jHBYrd9w"
        },
        {
            "url": "https://giphy.com/gifs/stream-famous-dollar-YT2dvOByEwXCdoYiA1"
        },
        {
            "url": "https://giphy.com/gifs/i-love-you-there-for-am-1BhGzgpZXYWwWMAGB1"
        },
        {
            "url": "https://giphy.com/gifs/life-like-twerk-9hlnWxjHqmH28"
        }
    ]
}


Préparez l'environnement



Commençons par préparer l'environnement.



Tout d'abord, nous devons créer un dossier de projet et un environnement virtuel: 



mkdir giphynav-aiohttp-tutorial
cd giphynav-aiohttp-tutorial
python3 -m venv venv


Maintenant, activons l'environnement virtuel: 



. venv/bin/activate


L'environnement est prĂȘt, commençons maintenant par la structure du projet.



Structure du projet



Dans cette section, nous organiserons la structure du projet.



Créons la structure suivante dans le dossier actuel. Laissez tous les fichiers vides pour le moment.



Structure initiale: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   └── views.py
├── venv/
└── requirements.txt


Installer les dépendances



Il est temps d'installer les dépendances. Nous utiliserons des packages comme celui-ci:



  • dependency-injector - framework d'injection de dĂ©pendances
  • aiohttp - framework web
  • aiohttp-devtools - une bibliothĂšque d'aide qui fournit un serveur pour le dĂ©veloppement de redĂ©marrage en direct
  • pyyaml - bibliothĂšque d'analyse des fichiers YAML, utilisĂ©e pour lire la configuration
  • pytest-aiohttp- bibliothĂšque d'aide pour tester les aiohttpapplications
  • pytest-cov - bibliothĂšque d'aide pour mesurer la couverture de code par des tests


Ajoutons les lignes suivantes au fichier requirements.txt: 



dependency-injector
aiohttp
aiohttp-devtools
pyyaml
pytest-aiohttp
pytest-cov


Et exécutez dans le terminal: 



pip install -r requirements.txt


Installez en plus httpie. Il s'agit d'un client HTTP en ligne de commande. Nous l'

utiliserons pour tester manuellement l'API.



Exécutons dans le terminal: 



pip install httpie


Les dépendances sont installées. Construisons maintenant une application minimale.



Application minimale



Dans cette section, nous allons créer une application minimale. Il aura un point de terminaison qui renverra une réponse vide.



Éditons views.py: 



"""Views module."""

from aiohttp import web


async def index(request: web.Request) -> web.Response:
    query = request.query.get('query', 'Dependency Injector')
    limit = int(request.query.get('limit', 10))

    gifs = []

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


Ajoutons maintenant un conteneur de dépendances (plus simplement un conteneur). Le conteneur contiendra tous les composants de l'application. Ajoutons les deux premiers composants. Ceci est une aiohttpapplication et une présentation index.



Éditons containers.py: 



"""Application containers module."""

from dependency_injector import containers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

    index_view = aiohttp.View(views.index)


Nous devons maintenant créer une fabrique d' aiohttpapplications. Il est généralement appelé

create_app(). Cela créera un conteneur. Le conteneur sera utilisé pour créer l' aiohttpapplication. La derniÚre étape consiste à configurer le routage - nous attribuerons une vue index_viewdu conteneur pour gérer les demandes à la racine de "/"notre application.



Éditons application.py: 



"""Application module."""

from aiohttp import web

from .containers import ApplicationContainer


def create_app():
    """Create and return aiohttp application."""
    container = ApplicationContainer()

    app: web.Application = container.app()
    app.container = container

    app.add_routes([
        web.get('/', container.index_view.as_view()),
    ])

    return app


Le conteneur est le premier objet de l'application. Il est utilisé pour récupérer tous les autres objets.


Nous sommes maintenant prĂȘts Ă  lancer notre application:



Exécutez la commande dans le terminal: 



adev runserver giphynavigator/application.py --livereload


La sortie devrait ressembler Ă  ceci: 



[18:52:59] Starting aux server at http://localhost:8001 ◆
[18:52:59] Starting dev server at http://localhost:8000 ●


Nous utilisons httpiepour vérifier le fonctionnement du serveur: 



http http://127.0.0.1:8000/


Tu verras: 



HTTP/1.1 200 OK
Content-Length: 844
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 21:01:50 GMT
Server: Python/3.8 aiohttp/3.6.2

{
    "gifs": [],
    "limit": 10,
    "query": "Dependency Injector"
}


L'application minimale est prĂȘte. Connectons l'API Giphy.



Client API Giphy



Dans cette section, nous allons intégrer notre application avec l'API Giphy. Nous créerons notre propre client API en utilisant le cÎté client aiohttp.



Créez un fichier vide giphy.pydans le package giphynavigator: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   └── views.py
├── venv/
└── requirements.txt


et ajoutez-y les lignes suivantes: 



"""Giphy client module."""

from aiohttp import ClientSession, ClientTimeout


class GiphyClient:

    API_URL = 'http://api.giphy.com/v1'

    def __init__(self, api_key, timeout):
        self._api_key = api_key
        self._timeout = ClientTimeout(timeout)

    async def search(self, query, limit):
        """Make search API call and return result."""
        if not query:
            return []

        url = f'{self.API_URL}/gifs/search'
        params = {
            'q': query,
            'api_key': self._api_key,
            'limit': limit,
        }
        async with ClientSession(timeout=self._timeout) as session:
            async with session.get(url, params=params) as response:
                if response.status != 200:
                    response.raise_for_status()
                return await response.json()


Nous devons maintenant ajouter le GiphyClient au conteneur. GiphyClient a deux dĂ©pendances qui doivent ĂȘtre transmises lors de sa crĂ©ation: la clĂ© API et le dĂ©lai d'expiration de la demande. Pour ce faire, nous devrons utiliser deux nouveaux fournisseurs du module dependency_injector.providers:



  • Le fournisseur FactorycrĂ©era le GiphyClient.
  • Le fournisseur Configurationenverra la clĂ© API et le dĂ©lai d'expiration au GiphyClient.


Éditons containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    index_view = aiohttp.View(views.index)


Nous avons utilisé les paramÚtres de configuration avant de définir leurs valeurs. C'est le principe selon lequel le fournisseur fonctionne Configuration.



Nous utilisons d'abord, puis nous définissons les valeurs.



Ajoutons maintenant le fichier de configuration.

Nous utiliserons YAML.



Créez un fichier vide config.ymlà la racine du projet: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   └── views.py
├── venv/
├── config.yml
└── requirements.txt


Et remplissez-le avec les lignes suivantes: 



giphy:
  request_timeout: 10


Nous utiliserons une variable d'environnement pour transmettre la clé API GIPHY_API_KEY .



Maintenant, nous devons éditer create_app()pour faire 2 actions lorsque l'application démarre:



  • Charger la configuration depuis config.yml
  • Charger la clĂ© API Ă  partir de la variable d'environnement GIPHY_API_KEY


Modifier application.py: 



"""Application module."""

from aiohttp import web

from .containers import ApplicationContainer


def create_app():
    """Create and return aiohttp application."""
    container = ApplicationContainer()
    container.config.from_yaml('config.yml')
    container.config.giphy.api_key.from_env('GIPHY_API_KEY')

    app: web.Application = container.app()
    app.container = container

    app.add_routes([
        web.get('/', container.index_view.as_view()),
    ])

    return app


Nous devons maintenant créer une clé API et la définir sur une variable d'environnement.



Afin de ne pas perdre de temps là-dessus, utilisez maintenant cette clé: 



export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0


Suivez ce tutoriel pour créer votre propre clé API Giphy .


La création et la configuration du client de l'API Giphy sont terminées. Passons au service de recherche.



Service de recherche



Il est temps d'ajouter un service de recherche SearchService. Il sera:



  • Chercher
  • Formater la rĂ©ponse reçue


SearchServiceutilisera GiphyClient.



Créez un fichier vide services.pydans le package giphynavigator: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   ├── services.py
│   └── views.py
├── venv/
└── requirements.txt


et ajoutez-y les lignes suivantes: 



"""Services module."""

from .giphy import GiphyClient


class SearchService:

    def __init__(self, giphy_client: GiphyClient):
        self._giphy_client = giphy_client

    async def search(self, query, limit):
        """Search for gifs and return formatted data."""
        if not query:
            return []

        result = await self._giphy_client.search(query, limit)

        return [{'url': gif['url']} for gif in result['data']]


Lors de la création, SearchServicevous devez transférer GiphyClient. Nous l'indiquerons lorsque nous l'ajouterons SearchServiceau conteneur.



Éditons containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

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

    index_view = aiohttp.View(views.index)


Le service de recherche est maintenant SearchServiceterminé. Dans la section suivante, nous allons le connecter à notre vue.



Connecter la recherche



Nous sommes maintenant prĂȘts pour que la recherche fonctionne. Utilisons SearchServiceen indexvue.



Modifier views.py: 



"""Views module."""

from aiohttp import web

from .services import SearchService


async def index(
        request: web.Request,
        search_service: SearchService,
) -> web.Response:
    query = request.query.get('query', 'Dependency Injector')
    limit = int(request.query.get('limit', 10))

    gifs = await search_service.search(query, limit)

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


Modifions maintenant le conteneur pour transmettre la dépendance SearchServiceà la vue indexlorsqu'elle est appelée.



Modifier containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

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

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
    )


Assurez-vous que l'application est en cours d'exécution ou exécutée: 



adev runserver giphynavigator/application.py --livereload


et faites une demande Ă  l'API dans le terminal: 



http http://localhost:8000/ query=="wow,it works" limit==5


Tu verras: 



HTTP/1.1 200 OK
Content-Length: 850
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 22:22:55 GMT
Server: Python/3.8 aiohttp/3.6.2

{
    "gifs": [
        {
            "url": "https://giphy.com/gifs/discoverychannel-nugget-gold-rush-rick-ness-KGGPIlnC4hr4u2s3pY"
        },
        {
            "url": "https://giphy.com/gifs/primevideoin-ll1hyBS2IrUPLE0E71"
        },
        {
            "url": "https://giphy.com/gifs/jackman-works-jackmanworks-l4pTgQoCrmXq8Txlu"
        },
        {
            "url": "https://giphy.com/gifs/cat-massage-at-work-l46CzMaOlJXAFuO3u"
        },
        {
            "url": "https://giphy.com/gifs/everwhatproductions-fun-christmas-3oxHQCI8tKXoeW4IBq"
        },
    ],
    "limit": 10,
    "query": "wow,it works"
}






La recherche fonctionne.



Un peu de refactoring



Notre vue indexcontient deux valeurs codées en dur:



  • Terme de recherche par dĂ©faut
  • Limite du nombre de rĂ©sultats


Faisons un peu de refactoring. Nous transférerons ces valeurs dans la configuration.



Modifier views.py: 



"""Views module."""

from aiohttp import web

from .services import SearchService


async def index(
        request: web.Request,
        search_service: SearchService,
        default_query: str,
        default_limit: int,
) -> web.Response:
    query = request.query.get('query', default_query)
    limit = int(request.query.get('limit', default_limit))

    gifs = await search_service.search(query, limit)

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


Maintenant, nous avons besoin que ces valeurs soient transmises sur appel. Mettons Ă  jour le conteneur.



Modifier containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

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

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
        default_query=config.search.default_query,
        default_limit=config.search.default_limit,
    )


Maintenant, mettons Ă  jour le fichier de configuration.



Modifier config.yml: 



giphy:
  request_timeout: 10
search:
  default_query: "Dependency Injector"
  default_limit: 10


Le refactoring est terminé. Nous avons rendu notre application plus propre en déplaçant les valeurs codées en dur dans la configuration.



Dans la section suivante, nous ajouterons quelques tests.



Ajouter des tests



Ce serait bien d'ajouter quelques tests. Faisons le. Nous utiliserons pytest et la couverture .



Créez un fichier vide tests.pydans le package giphynavigator: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   ├── services.py
│   ├── tests.py
│   └── views.py
├── venv/
└── requirements.txt


et ajoutez-y les lignes suivantes: 



"""Tests module."""

from unittest import mock

import pytest

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


@pytest.fixture
def app():
    return create_app()


@pytest.fixture
def client(app, aiohttp_client, loop):
    return loop.run_until_complete(aiohttp_client(app))


async def test_index(client, 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.container.giphy_client.override(giphy_client_mock):
        response = await client.get(
            '/',
            params={
                'query': 'test',
                'limit': 10,
            },
        )

    assert response.status == 200
    data = await 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(client, app):
    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 == 200
    data = await response.json()
    assert data['gifs'] == []


async def test_index_default_params(client, app):
    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 == 200
    data = await response.json()
    assert data['query'] == app.container.config.search.default_query()
    assert data['limit'] == app.container.config.search.default_limit()


Commençons maintenant les tests et vérifions la couverture: 



py.test giphynavigator/tests.py --cov=giphynavigator


Tu verras: 



platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1
plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0
collected 3 items

giphynavigator/tests.py ...                                     [100%]

---------- coverage: platform darwin, python 3.8.3-final-0 -----------
Name                            Stmts   Miss  Cover
---------------------------------------------------
giphynavigator/__init__.py          0      0   100%
giphynavigator/__main__.py          5      5     0%
giphynavigator/application.py      10      0   100%
giphynavigator/containers.py       10      0   100%
giphynavigator/giphy.py            16     11    31%
giphynavigator/services.py          9      1    89%
giphynavigator/tests.py            35      0   100%
giphynavigator/views.py             7      0   100%
---------------------------------------------------
TOTAL                              92     17    82%


Remarquez comment nous remplaçons giphy_client par mock en utilisant la méthode .override(). De cette façon, vous pouvez remplacer la valeur de retour de n'importe quel fournisseur.



Le travail est terminé. Maintenant, résumons.



Conclusion



Nous avons construit une aiohttpapplication API REST en utilisant le principe d'injection de dépendances. Nous avons utilisé Dependency Injector comme cadre d'injection de dépendances.



L'avantage que vous obtenez avec Dependency Injector est le conteneur.



Le conteneur commence Ă  porter ses fruits lorsque vous devez comprendre ou modifier la structure de votre application. Avec un conteneur, c'est facile car tous les composants de l'application et leurs dĂ©pendances se trouvent au mĂȘme endroit: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = aiohttp.Application(web.Application)

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

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
        default_query=config.search.default_query,
        default_limit=config.search.default_limit,
    )




Un conteneur comme carte de votre application. Vous savez toujours ce qui dépend de quoi.



Et aprĂšs?





More articles:


All Articles