我是Dependency Injector的创建者。这是Python的依赖项注入框架。
继续使用Dependency Injector来构建应用程序的一系列教程。
在本教程中，我想向您展示如何使用依赖注入器进行
aiohttp应用程序开发。
该手册包括以下部分：
可以在Github上找到完成的项目。
首先，您必须具备：
- Python 3.5+
- 虚拟环境
并且希望具有：
- aiohttp的初步开发技能
- 了解依赖注入的原理
我们要建造什么？
我们将构建一个REST API应用程序，以在Giphy上搜索有趣的gif文件。我们称之为Giphy Navigator。
Giphy Navigator如何工作？
- 客户端发送一个请求，指示要查找的内容以及返回多少结果。
- Giphy Navigator返回json响应。
- 答案包括：
- 搜索查询
- 结果数
- GIF网址清单
样本回复：
{
"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"
}
]
}
准备环境
让我们从准备环境开始。
首先，我们需要创建一个项目文件夹和一个虚拟环境：
mkdir giphynav-aiohttp-tutorial
cd giphynav-aiohttp-tutorial
python3 -m venv venv
现在让我们激活虚拟环境：
. venv/bin/activate
环境已经准备就绪，现在让我们从项目结构开始。
项目结构
在本节中，我们将组织项目的结构。
让我们在当前文件夹中创建以下结构。现在将所有文件留空。
初始结构：
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ └── views.py
├── venv/
└── requirements.txt
安装依赖
是时候安装依赖项了。我们将使用如下软件包：
dependency-injector-依赖注入框架aiohttp-网络框架aiohttp-devtools-为在线重启开发提供服务器的帮助程序库pyyaml-用于解析YAML文件的库，用于读取配置pytest-aiohttp-用于测试aiohttp应用程序的帮助程序库pytest-cov-帮助库，用于测试测试的代码覆盖率
让我们在文件中添加以下几行
requirements.txt：
dependency-injector
aiohttp
aiohttp-devtools
pyyaml
pytest-aiohttp
pytest-cov
并在终端中执行：
pip install -r requirements.txt
额外安装
httpie。这是一个命令行HTTP客户端。我们将
使用它来手动测试API。
让我们在终端中执行：
pip install httpie
依赖项已安装。现在让我们构建一个最小的应用程序。
最少的应用
在本节中，我们将构建一个最小的应用程序。它将具有一个端点，该端点将返回空响应。
让我们编辑
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,
},
)
现在，我们添加一个依赖项容器（还只是一个容器）。该容器将包含应用程序的所有组件。让我们添加前两个组件。这是一个
aiohttp应用程序和演示index。
让我们编辑
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)
现在我们需要创建一个
aiohttp应用程序工厂。通常称为
create_app()。它将创建一个容器。该容器将用于创建aiohttp应用程序。最后一步是设置路由-我们将从index_view容器中分配一个视图以处理"/"对应用程序根目录的请求。
让我们编辑
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
容器是应用程序中的第一个对象。它用于获取所有其他对象。
现在我们准备启动我们的应用程序：
在终端中运行命令：
adev runserver giphynavigator/application.py --livereload
输出应如下所示：
[18:52:59] Starting aux server at http://localhost:8001 ◆
[18:52:59] Starting dev server at http://localhost:8000 ●
我们
httpie用来检查服务器的运行情况：
http http://127.0.0.1:8000/
你会看见：
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"
}
最小的应用程序已准备就绪。让我们连接Giphy API。
Giphy API客户端
在本节中，我们将把我们的应用程序与Giphy API集成在一起。我们将使用客户端创建自己的API客户端
aiohttp。在包中
创建一个空文件：
giphy.pygiphynavigator
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ ├── giphy.py
│ └── views.py
├── venv/
└── requirements.txt
并添加以下几行：
"""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()
现在我们需要将GiphyClient添加到容器中。GiphyClient在创建时需要传递两个依赖关系：API密钥和请求超时。为此，我们将需要使用模块中的两个新提供程序
dependency_injector.providers：
- 提供者
Factory将创建GiphyClient。 - 提供者
Configuration会将API密钥和超时发送到GiphyClient。
让我们编辑
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)
在设置它们的值之前，我们使用了配置参数。这是提供者工作的原则Configuration。
首先我们使用，然后设置值。
现在，让我们添加配置文件。
我们将使用YAML。在项目的根目录下
创建一个空文件
config.yml：
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ ├── giphy.py
│ └── views.py
├── venv/
├── config.yml
└── requirements.txt
并用以下几行填写：
giphy:
request_timeout: 10
我们将使用环境变量来传递API密钥
GIPHY_API_KEY 。
现在，我们需要编辑
create_app()以在应用程序启动时执行2个操作：
- 从加载配置
config.yml - 从环境变量加载API密钥
GIPHY_API_KEY
编辑
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
现在，我们需要创建一个API密钥并将其设置为环境变量。
为了不浪费时间，现在使用以下密钥：
export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0
按照本教程创建自己的Giphy API密钥。
Giphy API客户端创建和配置设置已完成。让我们继续搜索服务。
搜索服务
现在该添加搜索服务了
SearchService。他会的：
- 搜索
- 格式化收到的回复
SearchService将使用GiphyClient。在包中
创建一个空文件：
services.pygiphynavigator
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ ├── giphy.py
│ ├── services.py
│ └── views.py
├── venv/
└── requirements.txt
并添加以下几行：
"""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']]
创建时，
SearchService您需要转移GiphyClient。当我们将其添加SearchService到容器中时，我们将进行指示。
让我们编辑
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)
搜索服务现已
SearchService完成。在下一节中，我们将其连接到视图。
连接搜索
现在我们可以开始搜索了。让我们
SearchService在index视图中使用。
编辑
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,
},
)
现在，让我们更改容器，
SearchService以便index在调用依赖项时将其传递给视图。
编辑
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,
)
确保应用程序正在运行或正在运行：
adev runserver giphynavigator/application.py --livereload
并向终端中的API发出请求：
http http://localhost:8000/ query=="wow,it works" limit==5
你会看见：
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"
}
搜索有效。
一点重构
我们的视图
index包含两个硬编码值：
- 默认搜索词
- 结果数限制
让我们做一些重构。我们会将这些值传输到配置中。
编辑
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,
},
)
现在我们需要这些值在调用时传递。让我们更新容器。
编辑
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,
)
现在，让我们更新配置文件。
编辑
config.yml：
giphy:
request_timeout: 10
search:
default_query: "Dependency Injector"
default_limit: 10
重构完成。通过将硬编码值移入配置，我们使应用程序更整洁。
在下一节中，我们将添加一些测试。
添加测试
添加一些测试会很好。我们开始做吧。我们将使用pytest和coverage。在包中
创建一个空文件：
tests.pygiphynavigator
./
├── giphynavigator/
│ ├── __init__.py
│ ├── application.py
│ ├── containers.py
│ ├── giphy.py
│ ├── services.py
│ ├── tests.py
│ └── views.py
├── venv/
└── requirements.txt
并添加以下几行：
"""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()
现在开始测试并检查覆盖范围：
py.test giphynavigator/tests.py --cov=giphynavigator
你会看见：
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%
注意我们如何giphy_client使用方法替换为模拟.override()。这样，您可以覆盖任何提供程序的返回值。
工作已经完成。现在让我们总结一下。
结论
我们已经
aiohttp使用依赖注入原理构建了一个REST API应用程序。我们使用了依赖注入器作为依赖注入框架。
Dependency Injector带来的好处是容器。
当您需要了解或更改应用程序的结构时，容器便开始发挥作用。使用容器，这很容易，因为应用程序的所有组件及其依赖项都位于一个位置：
"""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,
)
一个容器，作为您的应用程序的映射。您总是知道什么取决于什么。
下一步是什么？
- 在GitHub上了解有关Dependency Injector的更多信息
- 退房在阅读文档的文档
- 有疑问或发现错误？在Github上打开一个问题