Aiohttp + Dependency Injector - tutorial de inyección de dependencias

Hola,



soy el creador de Dependency Injector . Este es un marco de inyección de dependencia para Python.



Continuando con una serie de tutoriales sobre el uso de Dependency Injector para crear aplicaciones.



En este tutorial, quiero mostrarle cómo utilizar el inyector de dependencia para el desarrollo de aiohttpaplicaciones.



El manual consta de las siguientes partes:

1. ¿Qué vamos a construir?
2. Preparando el medio ambiente
3. Estructura del proyecto
4. Instalación de dependencias
5. Aplicación mínima
6. Cliente API de Giphy
7. Servicio de búsqueda
8. Conectar búsqueda
9. Un poco de refactorización
10. Agregar pruebas
11. Conclusión

El proyecto completo se puede encontrar en Github .



Para empezar debes tener:

• Python 3.5+
• Ambiente virtual

Y es deseable tener:

• Habilidades de desarrollo inicial con aiohttp
• Comprender el principio de inyección de dependencia

¿Qué vamos a construir?

Crearemos una aplicación API REST que busca gifs divertidos en Giphy . Llamémoslo Giphy Navigator.



¿Cómo funciona Giphy Navigator?

• El cliente envía una solicitud indicando qué buscar y cuántos resultados devolver.
• Giphy Navigator devuelve la respuesta json.
• La respuesta incluye:
  
  
  • consulta de busqueda
  • número de resultados
  • Lista de URL GIF

Respuesta de muestra:

{
    "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"
        }
    ]
}

Prepara el medio ambiente

Empecemos por preparar el medio ambiente.



En primer lugar, necesitamos crear una carpeta de proyecto y un entorno virtual:

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

Ahora activemos el entorno virtual:

. venv/bin/activate

El entorno está listo, ahora comencemos con la estructura del proyecto.

Estructura del proyecto

En esta sección, organizaremos la estructura del proyecto.



Creemos la siguiente estructura en la carpeta actual. Deje todos los archivos vacíos por ahora.



Estructura inicial:

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

Instalación de dependencias

Es hora de instalar las dependencias. Usaremos paquetes como este:

• dependency-injector - marco de inyección de dependencia
• aiohttp - marco web
• aiohttp-devtools - una biblioteca auxiliar que proporciona un servidor para el desarrollo de reinicio en vivo
• pyyaml - biblioteca para analizar archivos YAML, utilizada para leer la configuración
• pytest-aiohttp- biblioteca auxiliar para probar aiohttpaplicaciones
• pytest-cov - biblioteca auxiliar para medir la cobertura del código mediante pruebas

Agreguemos las siguientes líneas al archivo requirements.txt:

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

Y ejecutar en la terminal:

pip install -r requirements.txt

Instalar adicionalmente httpie. Es un cliente HTTP de línea de comandos. Lo

usaremos para probar manualmente la API.



Ejecutemos en la terminal:

pip install httpie

Las dependencias están instaladas. Ahora construyamos una aplicación mínima.

Aplicación mínima

En esta sección, crearemos una aplicación mínima. Tendrá un punto final que devolverá una respuesta vacía.



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

Ahora agreguemos un contenedor para dependencias (en adelante, solo un contenedor). El contenedor contendrá todos los componentes de la aplicación. Agreguemos los dos primeros componentes. Esta es una aiohttpaplicación y presentación index.



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

Ahora necesitamos crear una fábrica de aiohttpaplicaciones. Suele llamarse

create_app(). Creará un contenedor. El contenedor se utilizará para crear la aiohttpaplicación. El último paso es configurar el enrutamiento: asignaremos una vista index_viewdesde el contenedor para manejar las solicitudes a la raíz de "/"nuestra aplicación.



Editemos 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

El contenedor es el primer objeto de la aplicación. Se usa para obtener todos los demás objetos.

Ahora estamos listos para lanzar nuestra aplicación:



Ejecuta el comando en la terminal:

adev runserver giphynavigator/application.py --livereload

La salida debería verse así:

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

Usamos httpiepara verificar el funcionamiento del servidor:

http http://127.0.0.1:8000/

Ya verás:

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"
}

La aplicación mínima está lista. Conectemos la API de Giphy.

Cliente API de Giphy

En esta sección, integraremos nuestra aplicación con la API de Giphy. Crearemos nuestro propio cliente API usando el lado del cliente aiohttp.



Cree un archivo vacío giphy.pyen el paquete giphynavigator:

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

y agregue las siguientes líneas:

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

Ahora necesitamos agregar el GiphyClient al contenedor. GiphyClient tiene dos dependencias que deben pasarse al crearlo: clave API y tiempo de espera de solicitud. Para hacer esto, necesitaremos utilizar dos nuevos proveedores del módulo dependency_injector.providers:

• El proveedor Factorycreará el GiphyClient.
• El proveedor Configurationenviará la clave de API y el tiempo de espera al GiphyClient.

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

Usamos los parámetros de configuración antes de establecer sus valores. Este es el principio por el que trabaja el proveedor Configuration.



Primero usamos, luego establecemos los valores.

Ahora agreguemos el archivo de configuración.

Usaremos YAML.



Cree un archivo vacío config.ymlen la raíz del proyecto:

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

Y rellénelo con las siguientes líneas:

giphy:
  request_timeout: 10

Usaremos una variable de entorno para pasar la clave API GIPHY_API_KEY .



Ahora necesitamos editar create_app()para hacer 2 acciones cuando se inicia la aplicación:

• Cargar configuración desde config.yml
• Cargar la clave API desde la variable de entorno GIPHY_API_KEY

Editar 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

Ahora necesitamos crear una clave API y configurarla en una variable de entorno.



Para no perder el tiempo en esto, ahora use esta clave:

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Siga este tutorial para crear su propia clave API de Giphy .

Se completó la creación y configuración del cliente de la API de Giphy. Pasemos al servicio de búsqueda.

Servicio de búsqueda

Es hora de agregar un servicio de búsqueda SearchService. Él:

• Buscar
• Formato de respuesta recibida

SearchServiceutilizará GiphyClient.



Cree un archivo vacío services.pyen el paquete giphynavigator:

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

y agregue las siguientes líneas:

"""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']]

Al crear, SearchServicedebe transferir GiphyClient. Esto lo indicaremos cuando lo agreguemos SearchServiceal contenedor.



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

El servicio de búsqueda ahora está SearchServicecompleto. En la siguiente sección, lo conectaremos a nuestra vista.

Conectar búsqueda

Ahora estamos listos para que la búsqueda funcione. Vamos uso SearchServiceen la indexvista.



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

Ahora cambiemos el contenedor para pasar la dependencia SearchServicea la vista indexcuando se la llame.



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

Asegúrese de que la aplicación se esté ejecutando o se esté ejecutando:

adev runserver giphynavigator/application.py --livereload

y hacer una solicitud a la API en la terminal:

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

Ya verás:

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 búsqueda funciona.

Un poco de refactorización

Nuestra vista indexcontiene dos valores codificados:

• Término de búsqueda predeterminado
• Límite para el número de resultados

Hagamos una pequeña refactorización. Transferiremos estos valores a la configuración.



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

Ahora necesitamos que estos valores se pasen en la llamada. Actualicemos el contenedor.



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

Ahora actualice el archivo de configuración.



Editar config.yml:

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

La refactorización está completa. Hemos hecho que nuestra aplicación sea más limpia al mover valores codificados en la configuración.



En la siguiente sección, agregaremos algunas pruebas.

Agregar pruebas

Sería bueno agregar algunas pruebas. Vamos a hacerlo. Usaremos Pytest y Cobertura .



Cree un archivo vacío tests.pyen el paquete giphynavigator:

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

y agregue las siguientes líneas:

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

Ahora comencemos a probar y verificar la cobertura:

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

Ya verás:

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%

Observe cómo reemplazamos giphy_client con simulacro usando el método .override(). De esta forma, puede anular el valor de retorno de cualquier proveedor.

El trabajo está hecho. Ahora resumamos.

Conclusión

Hemos creado una aiohttpaplicación de API REST utilizando el principio de inyección de dependencia. Usamos Dependency Injector como un marco de inyección de dependencias.



La ventaja que obtiene con Dependency Injector es el contenedor.



El contenedor comienza a dar sus frutos cuando necesita comprender o cambiar la estructura de su aplicación. Con un contenedor, es fácil porque todos los componentes de la aplicación y sus dependencias están en un solo lugar:

"""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 contenedor como mapa de su aplicación. Siempre sabes qué depende de qué.

¿Que sigue?

• Obtenga más información sobre el inyector de dependencia en GitHub
• Consulte la documentación en Leer los documentos
• ¿Tiene alguna pregunta o encuentra un error? Abrir un problema en Github