Aiohttp + Dependency Injector - tutorial de injeção de dependência

Olá,



sou o criador do injetor de dependência . Esta é uma estrutura de injeção de dependência para Python.



Continuando uma série de tutoriais sobre como usar o Dependency Injector para construir aplicativos.



Neste tutorial, quero mostrar como usar o injetor de dependência para desenvolvimento de aiohttpaplicativos.



O manual consiste nas seguintes partes:

1. O que vamos construir?
2. Preparando o ambiente
3. Estrutura do projeto
4. Instalando dependências
5. Aplicação mínima
6. Cliente da API Giphy
7. Serviço de busca
8. Conectar pesquisa
9. Um pouco de refatoração
10. Adicionando testes
11. Conclusão

O projeto concluído pode ser encontrado no Github .



Para começar você deve ter:

• Python 3.5+
• Ambiente virtual

E é desejável ter:

• Habilidades de desenvolvimento inicial com aiohttp
• Compreender o princípio da injeção de dependência

O que vamos construir?

Estaremos construindo um aplicativo REST API que busca gifs engraçados no Giphy . Vamos chamá-lo de Giphy Navigator.



Como funciona o Giphy Navigator?

• O cliente envia uma solicitação indicando o que procurar e quantos resultados retornar.
• Giphy Navigator retorna uma resposta json.
• A resposta inclui:
  
  
  • consulta de pesquisa
  • número de resultados
  • Lista de urls GIF

Resposta de amostra:

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

Prepare o ambiente

Vamos começar preparando o ambiente.



Em primeiro lugar, precisamos criar uma pasta de projeto e um ambiente virtual:

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

Agora vamos ativar o ambiente virtual:

. venv/bin/activate

O ambiente está pronto, agora vamos começar com a estrutura do projeto.

Estrutura do projeto

Nesta seção, vamos organizar a estrutura do projeto.



Vamos criar a seguinte estrutura na pasta atual. Deixe todos os arquivos vazios por enquanto.



Estrutura inicial:

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

Instalando dependências

É hora de instalar as dependências. Usaremos pacotes como este:

• dependency-injector - framework de injeção de dependência
• aiohttp - estrutura da web
• aiohttp-devtools - uma biblioteca auxiliar que fornece um servidor para desenvolvimento de reinicialização ao vivo
• pyyaml - biblioteca para analisar arquivos YAML, usada para ler a configuração
• pytest-aiohttp- biblioteca auxiliar para aiohttpaplicativos de teste
• pytest-cov - biblioteca auxiliar para medir a cobertura de código por testes

Vamos adicionar as seguintes linhas ao arquivo requirements.txt:

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

E execute no terminal:

pip install -r requirements.txt

Instale adicionalmente httpie. É um cliente HTTP de linha de comando. Vamos

usá-lo para testar manualmente a API.



Vamos executar no terminal:

pip install httpie

As dependências são instaladas. Agora vamos construir um aplicativo mínimo.

Aplicação mínima

Nesta seção, construiremos um aplicativo mínimo. Ele terá um endpoint que retornará uma resposta vazia.



Vamos editar 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,
        },
    )

Agora vamos adicionar um contêiner para dependências (doravante apenas um contêiner). O contêiner conterá todos os componentes do aplicativo. Vamos adicionar os dois primeiros componentes. Este é um aiohttpaplicativo e uma apresentação index.



Vamos editar 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)

Agora precisamos criar uma fábrica de aiohttpaplicativos. Geralmente é chamado

create_app(). Isso criará um contêiner. O contêiner será usado para criar o aiohttpaplicativo. A etapa final é configurar o roteamento - atribuiremos uma visão index_viewdo contêiner para lidar com as solicitações à raiz do "/"nosso aplicativo.



Vamos editar 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

O contêiner é o primeiro objeto do aplicativo. É usado para obter todos os outros objetos.

Agora estamos prontos para lançar nosso aplicativo:



Execute o comando no terminal:

adev runserver giphynavigator/application.py --livereload

A saída deve ser semelhante a esta:

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

Usamos httpiepara verificar o funcionamento do servidor:

http http://127.0.0.1:8000/

Você verá:

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

O aplicativo mínimo está pronto. Vamos conectar a API Giphy.

Cliente da API Giphy

Nesta seção, iremos integrar nosso aplicativo com a API Giphy. Vamos criar nosso próprio cliente API usando o lado do cliente aiohttp.



Crie um arquivo vazio giphy.pyno pacote giphynavigator:

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

e adicione as seguintes linhas a ele:

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

Agora precisamos adicionar o GiphyClient ao contêiner. GiphyClient tem duas dependências que precisam ser passadas quando é criado: chave de API e tempo limite de solicitação. Para fazer isso, precisaremos usar dois novos provedores do módulo dependency_injector.providers:

• O provedor Factorycriará o GiphyClient.
• O provedor Configurationenviará a chave API e o tempo limite para o GiphyClient.

Vamos 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, 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 os parâmetros de configuração antes de definir seus valores. Este é o princípio pelo qual o provedor trabalha Configuration.



Primeiro usamos, depois definimos os valores.

Agora vamos adicionar o arquivo de configuração.

Usaremos YAML.



Crie um arquivo vazio config.ymlna raiz do projeto:

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

E preencha com as seguintes linhas:

giphy:
  request_timeout: 10

Usaremos uma variável de ambiente para passar a chave API GIPHY_API_KEY .



Agora precisamos editar create_app()para fazer 2 ações quando o aplicativo é iniciado:

• Carregar configuração de config.yml
• Carregar a chave API da variável de ambiente 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

Agora precisamos criar uma chave de API e defini-la como uma variável de ambiente.



Para não perder tempo com isso, agora use esta chave:

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Siga este tutorial para criar sua própria chave de API Giphy .

A criação do cliente Giphy API e a definição da configuração estão concluídas. Vamos passar para o serviço de pesquisa.

Serviço de busca

É hora de adicionar um serviço de pesquisa SearchService. Ele vai:

• Pesquisa
• Formatar resposta recebida

SearchServiceusará GiphyClient.



Crie um arquivo vazio services.pyno pacote giphynavigator:

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

e adicione as seguintes linhas a ele:

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

Ao criar, SearchServicevocê precisa transferir GiphyClient. Indicaremos isso quando o adicionarmos SearchServiceao contêiner.



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

O serviço de pesquisa agora está SearchServicecompleto. Na próxima seção, vamos conectá-lo à nossa visão.

Conectar pesquisa

Agora estamos prontos para que a pesquisa funcione. Vamos usar SearchServiceem 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,
        },
    )

Agora vamos alterar o contêiner para passar a dependência SearchServicepara a visualização indexquando for chamada.



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

Certifique-se de que o aplicativo esteja em execução ou execute:

adev runserver giphynavigator/application.py --livereload

e faça uma solicitação à API no terminal:

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

Você verá:

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

A pesquisa funciona.

Um pouco de refatoração

Nossa visualização indexcontém dois valores codificados:

• Termo de pesquisa padrão
• Limite para o número de resultados

Vamos refatorar um pouco. Vamos transferir esses valores para a configuração.



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

Agora precisamos que esses valores sejam repassados ​​à chamada. Vamos atualizar o contêiner.



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

Agora vamos atualizar o arquivo de configuração.



Editar config.yml:

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

A refatoração está completa. Tornamos nosso aplicativo mais limpo movendo os valores codificados para a configuração.



Na próxima seção, adicionaremos alguns testes.

Adicionando testes

Seria bom adicionar alguns testes. Vamos fazer isso. Estaremos usando teste e cobertura .



Crie um arquivo vazio tests.pyno pacote giphynavigator:

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

e adicione as seguintes linhas a ele:

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

Agora vamos começar a testar e verificar a cobertura:

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

Você verá:

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 como substituímos giphy_client por mock usando o método .override(). Dessa forma, você pode substituir o valor de retorno de qualquer provedor.

O trabalho está feito. Agora vamos resumir.

Conclusão

Construímos um aiohttpaplicativo REST API usando o princípio de injeção de dependência. Usamos o Dependency Injector como uma estrutura de injeção de dependência.



A vantagem que você obtém com o Dependency Injector é o contêiner.



O contêiner começa a pagar quando você precisa entender ou alterar a estrutura do seu aplicativo. Com um contêiner, é fácil porque todos os componentes do aplicativo e suas dependências estão em um só 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,
    )

Um contêiner como um mapa de seu aplicativo. Você sempre sabe o que depende do quê.

Qual é o próximo?

• Saiba mais sobre o Dependency Injector no GitHub
• Confira a documentação em Read the Docs
• Tem uma pergunta ou encontrou um bug? Abra um problema no Github