教程(ASGI)¶
在本教程中，我们将介绍如何为简单的图像共享服务构建API。在此过程中，我们将讨论异步Falcon应用程序的基本剖析：响应器、路由、中间件、在执行器中执行同步功能等！
备注
本教程介绍了Falcon的异步风格，它使用 ASGI 协议。
同步 (WSGI _)Falcon应用程序开发在我们的 WSGI tutorial 。
Falcon的新用户可能还想选择WSGI风格来熟悉Falcon的基本概念。
第一步¶
让我们首先创建一个全新的环境和相应的项目目录结构，如下所示 第一步 摘自WSGI教程：
asgilook
├── .venv
└── asgilook
├── __init__.py
└── app.py
我们将创建一个 virtualenv 使用 venv 来自标准库的模块(Falcon需要使用Python3.7+)：：
$ mkdir asgilook
$ python3 -m venv asgilook/.venv
$ source asgilook/.venv/bin/activate
备注
如果您的Python发行版碰巧没有包含 venv 模块，您可以随时安装和使用 virtualenv 取而代之的是。
小技巧
我们中的一些人觉得管理起来很方便 虚拟环境 s与 virtualenvwrapper 或 pipenv ，特别是当涉及到在几个环境之间跳跃时。
接下来，将Falcon安装到您的 虚拟环境 。ASGI支持需要3.0版或更高版本：：
$ pip install "falcon>=3.*"
然后，您可以创建一个基本的 Falcon ASGI application 通过添加一个 asgilook/app.py 包含以下内容的模块：
import falcon.asgi
app = falcon.asgi.App()
就像在 WSGI tutorial's introduction ，让我们不要忘了标记 asgilook 作为Python包：
$ touch asgilook/__init__.py
托管我们的应用程序¶
要运行我们的异步应用程序，我们需要一个 ASGI 应用程序服务器。受欢迎的选择包括：
对于像我们这样的简单教程应用程序，上面的任何一个都可以。让我们挑选最受欢迎的 uvicorn 目前：：
$ pip install uvicorn
另请参阅： ASGI Server Installation 。
既然我们做了，我们就把便携的 HTTPie 帮助我们锻炼应用程序的HTTP客户端：：
$ pip install httpie
现在，让我们尝试加载我们的应用程序：：
$ uvicorn asgilook.app:app
INFO: Started server process [2020]
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Waiting for application startup.
INFO: Application startup complete.
我们可以通过尝试访问上面提供的URL来验证它是否正常工作 uvicorn ：：
$ http http://127.0.0.1:8000
HTTP/1.1 404 Not Found
content-length: 0
content-type: application/json
date: Sun, 05 Jul 2020 13:37:01 GMT
server: uvicorn
喔呼，成功了！
嗯，算是吧。继续添加一些真正的功能！
配置¶
接下来，让我们通过允许用户修改存储图像的文件系统路径来使我们的应用程序可配置。我们还将允许自定义UUID生成器。
由于Falcon没有规定具体的配置库或策略，因此我们可以自由选择自己的冒险(另请参阅我们常见问题解答中的相关问题： 推荐的应用程序配置方法是什么？ )。
在本教程中，我们将只传递一个 Config 实例映射到资源初始值设定项，以便于测试(将在本教程后面介绍)。创建一个新模块， config.py 紧邻 app.py ，并向其中添加以下代码：
import os
import pathlib
import uuid
class Config:
DEFAULT_CONFIG_PATH = '/tmp/asgilook'
DEFAULT_UUID_GENERATOR = uuid.uuid4
def __init__(self):
self.storage_path = pathlib.Path(
os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH))
self.storage_path.mkdir(parents=True, exist_ok=True)
self.uuid_generator = Config.DEFAULT_UUID_GENERATOR
图像存储¶
由于我们要读写图像文件，因此必须小心避免在I/O期间阻塞应用程序。 aiofiles A尝试：：
pip install aiofiles
另外，让我们稍微改变一下原来的WSGI“外观”设计，把所有上传的图片都转换成JPEG格式 Pillow 库：：
pip install Pillow
我们现在可以实现一个基本的异步映像存储。将以下代码另存为 store.py 紧邻 app.py 和 config.py ：
import asyncio
import datetime
import io
import aiofiles
import falcon
import PIL.Image
class Image:
def __init__(self, config, image_id, size):
self._config = config
self.image_id = image_id
self.size = size
self.modified = datetime.datetime.utcnow()
@property
def path(self):
return self._config.storage_path / self.image_id
@property
def uri(self):
return f'/images/{self.image_id}.jpeg'
def serialize(self):
return {
'id': self.image_id,
'image': self.uri,
'modified': falcon.dt_to_http(self.modified),
'size': self.size,
}
class Store:
def __init__(self, config):
self._config = config
self._images = {}
def _load_from_bytes(self, data):
return PIL.Image.open(io.BytesIO(data))
def _convert(self, image):
rgb_image = image.convert('RGB')
converted = io.BytesIO()
rgb_image.save(converted, 'JPEG')
return converted.getvalue()
def get(self, image_id):
return self._images.get(image_id)
def list_images(self):
return sorted(self._images.values(), key=lambda item: item.modified)
async def save(self, image_id, data):
loop = asyncio.get_running_loop()
image = await loop.run_in_executor(None, self._load_from_bytes, data)
converted = await loop.run_in_executor(None, self._convert, image)
path = self._config.storage_path / image_id
async with aiofiles.open(path, 'wb') as output:
await output.write(converted)
stored = Image(self._config, image_id, image.size)
self._images[image_id] = stored
return stored
在这里，我们使用以下命令存储数据 aiofiles ，然后运行 Pillow 默认情况下的图像变换函数 ThreadPoolExecutor ，希望这些图像操作中至少有一部分在处理过程中释放GIL。
备注
这个 ProcessPoolExecutor 是不发布GIL的长时间运行任务的另一种选择，例如CPU绑定的纯Python代码。但是，请注意， ProcessPoolExecutor 构建在 multiprocessing 模块，因此继承了它的警告：更高的同步开销，以及任务及其参数必须是可拾取的(这还意味着任务必须可以从全局命名空间访问，即匿名 lambda 根本行不通)。
图像资源¶
在Falcon的ASGI风格中，所有响应器方法、挂钩和中间件方法都必须是可等待的协程。让我们通过实现一个资源来同时表示单个图像和一组图像，看看这是如何工作的。将下面的代码放在一个名为的文件中 images.py ：
import aiofiles
import falcon
class Images:
def __init__(self, config, store):
self._config = config
self._store = store
async def on_get(self, req, resp):
resp.media = [image.serialize() for image in self._store.list_images()]
async def on_get_image(self, req, resp, image_id):
# NOTE: image_id: UUID is converted back to a string identifier.
image = self._store.get(str(image_id))
resp.stream = await aiofiles.open(image.path, 'rb')
resp.content_type = falcon.MEDIA_JPEG
async def on_post(self, req, resp):
data = await req.stream.read()
image_id = str(self._config.uuid_generator())
image = await self._store.save(image_id, data)
resp.location = image.uri
resp.media = image.serialize()
resp.status = falcon.HTTP_201
此模块是Falcon“资源”类的示例，如中所述 路由 。Falcon使用基于资源的路由来鼓励REST风格的架构。对于资源支持的每个HTTP方法，目标资源类只实现名称以 on_ 并且以小写的HTTP方法名结束(例如， on_get() ， on_patch() ， on_delete() 等)
备注
如果对于给定的HTTP谓词省略了Python方法，则框架将自动响应为 405 Method Not Allowed 。Falcon还为OPTIONS请求提供默认响应器，该响应器会考虑为目标资源实现哪些方法。
在这里，我们选择实现对单个映像的支持(它支持 GET 用于下载图像)和图像集合(其支持 GET 用于列出集合，以及 POST 用于上传新图像)位于相同的Falcon资源类中。为了实现这一点，Falcon路由器需要一种方法来确定调用哪些方法来获取集合，而不是调用条目。这是通过使用后缀路由完成的，如中所述 add_route() (另请参阅： 如何为同一资源实现过账和获取项目？ )。
或者，我们可以拆分实现以严格表示每个类一个基于REST的资源。在这种情况下，就不需要使用带后缀的响应器。根据应用程序的不同，使用两个类而不是一个类可能会导致更干净的设计。(另请参阅： 将相关路由映射到资源类的建议方法是什么？ )
备注
在此示例中，我们通过简单地将打开的 aiofiles 文件到 resp.stream 。这之所以有效，是因为Falcon包括对类似异步文件的对象进行流式处理的特殊处理。
警告
在生产部署中，直接从Web服务器(而不是通过Falcon ASGI应用程序)提供文件可能会更高效，因此应该首选。另请参阅： Falcon能提供静态文件吗？
同样值得注意的是， on_get_image() 应答者将收到 image_id 类型的 UUID 。这是怎么回事？这将会是怎样的呢？ image_id 与字符串路径段匹配的字段将成为 UUID ？
Falcon的默认路由器支持简单的验证和转换，使用 field converters 。在此示例中，我们将使用 UUIDConverter 要验证 image_id 输入为 UUID 。通过将其 shorthand identifiers 在路由的URI模板中；例如，对应于 on_get_image() 我将使用以下模板(另请参阅下一章以及 路由 )：
/images/{image_id:uuid}.jpeg
由于我们的应用程序在内部仍然以字符串标识符为中心，因此可以随意尝试重构图像 Store 要使用 UUID 这是天生的！
(或者，可以实现 custom field converter 要使用 uuid (仅用于验证，但返回未修改的字符串。)
备注
与异步构建块(响应器、中间件、挂钩等)形成对比在Falcon ASGI应用程序中，现场转换器是简单的同步数据转换功能，不需要执行任何I/O。
运行我们的应用程序¶
现在，我们已经准备好为应用程序配置路由，以将请求URL中的图像路径映射到资源类的一个实例。
让我们也重构我们的 app.py 模块，让我们调用 create_app() 在我们需要的任何地方。稍后当我们开始编写测试用例时，这将变得有用。
修改 app.py 按如下方式阅读：
import falcon.asgi
from .config import Config
from .images import Images
from .store import Store
def create_app(config=None):
config = config or Config()
store = Store(config)
images = Images(config, store)
app = falcon.asgi.App()
app.add_route('/images', images)
app.add_route('/images/{image_id:uuid}.jpeg', images, suffix='image')
return app
如前所述，我们需要将路由后缀用于 Images 类来区分单个图像的GET和整个图像集合的GET。
在这里，我们将 '/images/{{image_id:uuid}}.jpeg' URI模板指向单个图像资源。通过指定一个 'image' 后缀，我们使框架查找名称以 '_image' (例如， on_get_image() )。
我们还指定了 uuid 如上一节所讨论的现场转换器。
为了引导ASGI应用程序实例 uvicorn 作为参考，我们将创建一个简单的 asgi.py 包含以下内容的模块：
from .app import create_app
app = create_app()
运行应用程序与前面的命令行没有太大不同：：
$ uvicorn asgilook.asgi:app
提供了 uvicorn 按照上面的命令行启动，让我们尝试在单独的终端上传一些图像(更改下面的图片路径以指向现有文件)：
$ http POST localhost:8000/images @/home/user/Pictures/test.png
HTTP/1.1 201 Created
content-length: 173
content-type: application/json
date: Tue, 24 Dec 2019 17:32:18 GMT
location: /images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg
server: uvicorn
{
"id": "5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c",
"image": "/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg",
"modified": "Tue, 24 Dec 2019 17:32:19 GMT",
"size": [
462,
462
]
}
接下来，尝试检索上传的图像：：
$ http localhost:8000/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg
HTTP/1.1 200 OK
content-type: image/jpeg
date: Tue, 24 Dec 2019 17:34:53 GMT
server: uvicorn
transfer-encoding: chunked
+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+
我们还可以在Web浏览器中打开链接或通过管道将其传送给图像查看器，以验证图像是否已成功转换为JPEG。
现在让我们检查一下图像集合：：
$ http localhost:8000/images
HTTP/1.1 200 OK
content-length: 175
content-type: application/json
date: Tue, 24 Dec 2019 17:36:31 GMT
server: uvicorn
[
{
"id": "5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c",
"image": "/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg",
"modified": "Tue, 24 Dec 2019 17:32:19 GMT",
"size": [
462,
462
]
}
]
应用程序文件布局现在应该如下所示：：
asgilook
├── .venv
└── asgilook
├── __init__.py
├── app.py
├── asgi.py
├── config.py
├── images.py
└── store.py
动态缩略图¶
让我们假设我们的影像服务客户想要以多种分辨率呈现图像，例如， srcset 用于响应HTML图像或其他目的。
让我们添加一个新方法 Store.make_thumbnail() 要在飞翔上进行伸缩，请执行以下操作：
async def make_thumbnail(self, image, size):
async with aiofiles.open(image.path, 'rb') as img_file:
data = await img_file.read()
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, self._resize, data, size)
我们还将添加一个内部帮助器来运行 Pillow 再次将缩略图操作卸载到线程池执行器，希望Pillow可以释放GIL来执行某些操作：
def _resize(self, data, size):
image = PIL.Image.open(io.BytesIO(data))
image.thumbnail(size)
resized = io.BytesIO()
image.save(resized, 'JPEG')
return resized.getvalue()
这个 store.Image 类可以扩展为还将URI返回到缩略图：
def thumbnails(self):
def reductions(size, min_size):
width, height = size
factor = 2
while width // factor >= min_size and height // factor >= min_size:
yield (width // factor, height // factor)
factor *= 2
return [
f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'
for width, height in reductions(
self.size, self._config.min_thumb_size)]
在这里，我们只为一系列缩小的分辨率生成URI。实际伸缩将在请求这些资源时在飞翔上进行。
该系列中的每个缩略图的大小约为前一个缩略图的一半(四分之一区域)，类似于 mipmapping 在计算机图形学领域工作。您可能希望试验此分辨率分布。
更新后 store.py ，模块现在应该如下所示：
import asyncio
import datetime
import io
import aiofiles
import falcon
import PIL.Image
class Image:
def __init__(self, config, image_id, size):
self._config = config
self.image_id = image_id
self.size = size
self.modified = datetime.datetime.utcnow()
@property
def path(self):
return self._config.storage_path / self.image_id
@property
def uri(self):
return f'/images/{self.image_id}.jpeg'
def serialize(self):
return {
'id': self.image_id,
'image': self.uri,
'modified': falcon.dt_to_http(self.modified),
'size': self.size,
'thumbnails': self.thumbnails(),
}
def thumbnails(self):
def reductions(size, min_size):
width, height = size
factor = 2
while width // factor >= min_size and height // factor >= min_size:
yield (width // factor, height // factor)
factor *= 2
return [
f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'
for width, height in reductions(self.size, self._config.min_thumb_size)
]
class Store:
def __init__(self, config):
self._config = config
self._images = {}
def _load_from_bytes(self, data):
return PIL.Image.open(io.BytesIO(data))
def _convert(self, image):
rgb_image = image.convert('RGB')
converted = io.BytesIO()
rgb_image.save(converted, 'JPEG')
return converted.getvalue()
def _resize(self, data, size):
image = PIL.Image.open(io.BytesIO(data))
image.thumbnail(size)
resized = io.BytesIO()
image.save(resized, 'JPEG')
return resized.getvalue()
def get(self, image_id):
return self._images.get(image_id)
def list_images(self):
return sorted(self._images.values(), key=lambda item: item.modified)
async def make_thumbnail(self, image, size):
async with aiofiles.open(image.path, 'rb') as img_file:
data = await img_file.read()
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, self._resize, data, size)
async def save(self, image_id, data):
loop = asyncio.get_running_loop()
image = await loop.run_in_executor(None, self._load_from_bytes, data)
converted = await loop.run_in_executor(None, self._convert, image)
path = self._config.storage_path / image_id
async with aiofiles.open(path, 'wb') as output:
await output.write(converted)
stored = Image(self._config, image_id, image.size)
self._images[image_id] = stored
return stored
此外，设置最小分辨率是可行的，因为在非常小的缩略图(每个缩略图只有几千字节)之间切换的任何潜在好处都可能会被请求开销所掩盖。正如您在上面的代码片段中可能已经注意到的那样，我们将这个较低的大小限制引用为 self._config.min_thumb_size 。这个 app configuration 将需要更新以添加 min_thumb_size 选项(默认情况下初始化为64像素)，如下所示：
import os
import pathlib
import uuid
class Config:
DEFAULT_CONFIG_PATH = '/tmp/asgilook'
DEFAULT_MIN_THUMB_SIZE = 64
DEFAULT_UUID_GENERATOR = uuid.uuid4
def __init__(self):
self.storage_path = pathlib.Path(
os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH))
self.storage_path.mkdir(parents=True, exist_ok=True)
self.uuid_generator = Config.DEFAULT_UUID_GENERATOR
self.min_thumb_size = self.DEFAULT_MIN_THUMB_SIZE
让我们再添加一个 Thumbnails 资源来公开新功能。的最终版本 images.py 读取：
import aiofiles
import falcon
class Images:
def __init__(self, config, store):
self._config = config
self._store = store
async def on_get(self, req, resp):
resp.media = [image.serialize() for image in self._store.list_images()]
async def on_get_image(self, req, resp, image_id):
# NOTE: image_id: UUID is converted back to a string identifier.
image = self._store.get(str(image_id))
if not image:
raise falcon.HTTPNotFound
resp.stream = await aiofiles.open(image.path, 'rb')
resp.content_type = falcon.MEDIA_JPEG
async def on_post(self, req, resp):
data = await req.stream.read()
image_id = str(self._config.uuid_generator())
image = await self._store.save(image_id, data)
resp.location = image.uri
resp.media = image.serialize()
resp.status = falcon.HTTP_201
class Thumbnails:
def __init__(self, store):
self._store = store
async def on_get(self, req, resp, image_id, width, height):
image = self._store.get(str(image_id))
if not image:
raise falcon.HTTPNotFound
if req.path not in image.thumbnails():
raise falcon.HTTPNotFound
resp.content_type = falcon.MEDIA_JPEG
resp.data = await self._store.make_thumbnail(image, (width, height))
备注
尽管我们只构建了一个示例应用程序，但培养一种习惯，使您的代码在设计上和默认情况下都是安全的，这是一个好主意。
在这种情况下，我们可以看到，基于URI中嵌入的任意维度在飞翔上生成缩略图很容易被滥用来创建拒绝服务攻击。
通过对照允许值列表验证输入(在本例中为请求的路径)，可以缓解此特定攻击。
最后，一个新的缩略图 route 需要添加到 app.py 。这一步留给读者作为练习。
小技巧
从缩略图URI格式字符串中获取灵感：
f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'
缩略图路由的实际URI模板应该与上面的内容非常相似。
请记住，我们希望使用 uuid 用于 image_id 字段和图像尺寸 (width 和 height )理想情况下应转换为 int s.
(如果您遇到困难，请参阅最终版本的 app.py 在本教程的后面部分。)
备注
如果您尝试请求不存在的资源(例如，由于缺少路由，或者仅仅是URI中的拼写错误)，框架将自动呈现 HTTP 404 Not Found 通过引发 HTTPNotFound (除非该异常被 custom error handler 或者如果该路径与接收器前缀匹配)。
相反，如果路由与资源匹配，但没有相关HTTP方法的响应方，则Falcon将呈现 HTTP 405 Method Not Allowed 通过 HTTPMethodNotAllowed 。
新的 thumbnails 端点现在应该在飞翔上渲染缩略图：：
$ http POST localhost:8000/images @/home/user/Pictures/test.png
HTTP/1.1 201 Created
content-length: 319
content-type: application/json
date: Tue, 24 Dec 2019 18:58:20 GMT
location: /images/f2375273-8049-4b10-b17e-8851db9ac7af.jpeg
server: uvicorn
{
"id": "f2375273-8049-4b10-b17e-8851db9ac7af",
"image": "/images/f2375273-8049-4b10-b17e-8851db9ac7af.jpeg",
"modified": "Tue, 24 Dec 2019 18:58:21 GMT",
"size": [
462,
462
],
"thumbnails": [
"/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/231x231.jpeg",
"/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/115x115.jpeg"
]
}
$ http localhost:8000/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/115x115.jpeg
HTTP/1.1 200 OK
content-length: 2985
content-type: image/jpeg
date: Tue, 24 Dec 2019 19:00:14 GMT
server: uvicorn
+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+
同样，我们还可以在支持HTTP输入的浏览器或图像查看器中验证缩略图URI。
缓存响应¶
虽然在飞翔上缩放缩略图听起来很酷，我们也避免了很多讨厌的小文件散落在我们的存储中，但它消耗了cpu资源，而且我们很快就会发现我们的应用程序在负载下崩溃了。
让我们通过响应缓存来缓解这个问题。我们将使用Redis，利用 aioredis 对于异步支持，请执行以下操作：
pip install aioredis
我们还需要序列化响应数据( Content-Type 第一个版本中的表头和表体)； msgpack 应执行以下操作：
pip install msgpack
我们的应用程序显然需要访问Redis服务器。除了在您的机器上安装Redis服务器之外，您还可以：
在Docker中启动Redis，例如：：
docker run -p 6379:6379 redis
假设机器上安装了Redis，用户也可以尝试 pifpaf 只是为了暂时启动Redis
uvicorn：：pifpaf run redis -- uvicorn asgilook.asgi:app
我们将使用Falcon执行缓存 中间件 组件。同样，请注意，所有中间件回调都必须是异步的。即使是打电话 ping() 和 close() 在Redis连接上必须是 await Ed.但是我们怎么能 await 同步化内部的协同例程 create_app() 功能呢？
ASGI application lifespan events 来救援吧。ASGI应用程序服务器在应用程序启动和关闭时发出这些事件。
让我们实现 process_startup() 和 process_shutdown() 中间件中的处理程序分别在应用程序启动和关闭时执行代码：
async def process_startup(self, scope, event):
await self._redis.ping()
async def process_shutdown(self, scope, event):
await self._redis.close()
警告
寿命协议是一个可选的扩展；请检查您选择的ASGI服务器是否实施了该协议。
uvicorn (我们为本教程选择的)支持寿命。
至少，我们的中间件需要知道要使用的Redis主机。让我们还使我们的Redis连接工厂可配置，以便为生产和测试注入不同的Redis客户端实现。
备注
可以使用包装器方法隐式引用，而不是要求调用者将主机传递给连接工厂 self.redis_host 。事实证明，对于需要在多个地方创建客户端连接的应用程序，这样的设计可能会很有帮助。
假设我们把新的 configuration 项目 redis_host 和 redis_from_url() 分别为的最终版本 config.py 现在阅读：
import os
import pathlib
import uuid
import aioredis
class Config:
DEFAULT_CONFIG_PATH = '/tmp/asgilook'
DEFAULT_MIN_THUMB_SIZE = 64
DEFAULT_REDIS_HOST = 'redis://localhost'
DEFAULT_REDIS_FROM_URL = aioredis.from_url
DEFAULT_UUID_GENERATOR = uuid.uuid4
def __init__(self):
self.storage_path = pathlib.Path(
os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH)
)
self.storage_path.mkdir(parents=True, exist_ok=True)
self.redis_from_url = Config.DEFAULT_REDIS_FROM_URL
self.min_thumb_size = self.DEFAULT_MIN_THUMB_SIZE
self.redis_host = self.DEFAULT_REDIS_HOST
self.uuid_generator = Config.DEFAULT_UUID_GENERATOR
让我们通过实现另外两个中间件方法来完成Redis缓存组件，除了 process_startup() 和 process_shutdown() 。创建 cache.py 包含以下代码的模块：
import msgpack
class RedisCache:
PREFIX = 'asgilook:'
INVALIDATE_ON = frozenset({'DELETE', 'POST', 'PUT'})
CACHE_HEADER = 'X-ASGILook-Cache'
TTL = 3600
def __init__(self, config):
self._config = config
self._redis = self._config.redis_from_url(self._config.redis_host)
async def _serialize_response(self, resp):
data = await resp.render_body()
return msgpack.packb([resp.content_type, data], use_bin_type=True)
def _deserialize_response(self, resp, data):
resp.content_type, resp.data = msgpack.unpackb(data, raw=False)
resp.complete = True
resp.context.cached = True
async def process_startup(self, scope, event):
await self._redis.ping()
async def process_shutdown(self, scope, event):
await self._redis.close()
async def process_request(self, req, resp):
resp.context.cached = False
if req.method in self.INVALIDATE_ON:
return
key = f'{self.PREFIX}/{req.path}'
data = await self._redis.get(key)
if data is not None:
self._deserialize_response(resp, data)
resp.set_header(self.CACHE_HEADER, 'Hit')
else:
resp.set_header(self.CACHE_HEADER, 'Miss')
async def process_response(self, req, resp, resource, req_succeeded):
if not req_succeeded:
return
key = f'{self.PREFIX}/{req.path}'
if req.method in self.INVALIDATE_ON:
await self._redis.delete(key)
elif not resp.context.cached:
data = await self._serialize_response(resp)
await self._redis.set(key, data, ex=self.TTL)
要使缓存生效，我们还需要修改 app.py 要添加 RedisCache 组件添加到我们的应用程序的中间件列表中。的最终版本 app.py 应该如下所示：
import falcon.asgi
from .cache import RedisCache
from .config import Config
from .images import Images, Thumbnails
from .store import Store
def create_app(config=None):
config = config or Config()
cache = RedisCache(config)
store = Store(config)
images = Images(config, store)
thumbnails = Thumbnails(store)
app = falcon.asgi.App(middleware=[cache])
app.add_route('/images', images)
app.add_route('/images/{image_id:uuid}.jpeg', images, suffix='image')
app.add_route(
'/thumbnails/{image_id:uuid}/{width:int}x{height:int}.jpeg', thumbnails
)
return app
现在，后续访问 /thumbnails 应缓存，如 x-asgilook-cache 标题：：
$ http localhost:8000/thumbnails/167308e4-e444-4ad9-88b2-c8751a4e37d4/115x115.jpeg
HTTP/1.1 200 OK
content-length: 2985
content-type: image/jpeg
date: Tue, 24 Dec 2019 19:46:51 GMT
server: uvicorn
x-asgilook-cache: Hit
+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+
备注
留给读者的另一个练习是：单个图像直接从 aiofiles 实例，因此缓存目前对它们不起作用。
项目的结构现在应该如下所示：：
asgilook
├── .venv
└── asgilook
├── __init__.py
├── app.py
├── asgi.py
├── cache.py
├── config.py
├── images.py
└── store.py
测试我们的应用程序¶
到现在为止还好?我们只通过手动发送几个请求来测试我们的应用程序。我们测试了所有代码路径吗？我们是否涵盖了应用程序的典型用户输入？
创建全面的测试套件不仅对于验证应用程序目前的行为是否正确至关重要，而且对于限制随时间推移引入代码库的任何回归的影响也是至关重要的。
为了实现测试套件，我们将需要修改依赖项，并确定我们想要的抽象级别：
我们会运行真正的Redis服务器吗？
我们是将“真正的”文件存储在文件系统上，还是仅仅为
aiofiles？我们将注入真正的依赖项，还是使用模拟和猴子修补？
这里没有对与错，因为不同的测试策略(或其组合)在测试运行时间、实现新测试的难易程度、测试环境与生产的相似性等方面有各自的优势。
另一件可以选择的事情是测试框架。就像在 WSGI tutorial ，让我们使用 pytest 。这是一个品味问题；如果您更喜欢xUnit/JUnit风格的布局，那么您会对stdlib的 unittest 。
为了更快地交付有效的解决方案，我们将允许我们的测试访问实际的文件系统。为了我们的方便， pytest 提供几个开箱即用的临时目录实用程序。我们把它包起来吧 tmpdir_factory 要创建简单的 storage_path 我们将在套件中的所有测试之间共享的fixture(在 pytest 用“会话”作用域的装置来描述)。
小技巧
这个 pytest website includes in-depth documentation on the use of fixtures. Please visit pytest fixtures: explicit, modular, scalable 了解更多信息。
正如在 previous section ，有许多方法可以启动临时或永久的Redis服务器；或者完全模拟它。对于我们的测试，我们会试一试 fakeredis ，这是专门为编写单元测试量身定做的纯Python实现。
pytest 和 fakeredis 可以安装为：：
$ pip install fakeredis pytest
我们还将为我们的测试创建一个目录，并将其作为Python包，以避免导入本地实用程序模块或检查代码覆盖率时出现的任何问题：
$ mkdir -p tests
$ touch tests/__init__.py
接下来，让我们实现要替换的装置 uuid 和 aioredis ，并通过以下方式将它们注入到我们的测试中 conftest.py (将您的代码放在新创建的 tests 目录)：
import io
import random
import uuid
import fakeredis.aioredis
import falcon.asgi
import falcon.testing
import PIL.Image
import PIL.ImageDraw
import pytest
from asgilook.app import create_app
from asgilook.config import Config
@pytest.fixture()
def predictable_uuid():
fixtures = (
uuid.UUID('36562622-48e5-4a61-be67-e426b11821ed'),
uuid.UUID('3bc731ac-8cd8-4f39-b6fe-1a195d3b4e74'),
uuid.UUID('ba1c4951-73bc-45a4-a1f6-aa2b958dafa4'),
)
def uuid_func():
try:
return next(fixtures_it)
except StopIteration:
return uuid.uuid4()
fixtures_it = iter(fixtures)
return uuid_func
@pytest.fixture(scope='session')
def storage_path(tmpdir_factory):
return tmpdir_factory.mktemp('asgilook')
@pytest.fixture
def client(predictable_uuid, storage_path):
# NOTE(vytas): Unlike the sync FakeRedis, fakeredis.aioredis.FakeRedis
# seems to share a global state in 2.17.0 (by design or oversight).
# Make sure we initialize a new fake server for every test case.
def fake_redis_from_url(*args, **kwargs):
server = fakeredis.FakeServer()
return fakeredis.aioredis.FakeRedis(server=server)
config = Config()
config.redis_from_url = fake_redis_from_url
config.redis_host = 'redis://localhost'
config.storage_path = storage_path
config.uuid_generator = predictable_uuid
app = create_app(config)
return falcon.testing.TestClient(app)
@pytest.fixture(scope='session')
def png_image():
image = PIL.Image.new('RGBA', (640, 360), color='black')
draw = PIL.ImageDraw.Draw(image)
for _ in range(32):
x0 = random.randint(20, 620)
y0 = random.randint(20, 340)
x1 = random.randint(20, 620)
y1 = random.randint(20, 340)
if x0 > x1:
x0, x1 = x1, x0
if y0 > y1:
y0, y1 = y1, y0
draw.ellipse([(x0, y0), (x1, y1)], fill='yellow', outline='red')
output = io.BytesIO()
image.save(output, 'PNG')
return output.getvalue()
@pytest.fixture(scope='session')
def image_size():
def report_size(data):
image = PIL.Image.open(io.BytesIO(data))
return image.size
return report_size
备注
在 png_image 在上面的装置中，我们正在绘制随机图像，这些图像在每次运行测试时看起来都会有所不同。
如果您的测试流程能够做到这一点，那么在您的测试输入中引入一些不可预测性通常是一个很好的想法。这将提供更多的信心，使您的应用程序能够处理范围更广的输入，而不仅仅是专门为该唯一目的而设计的2-3个测试用例。
另一方面，随机输入可能会使断言变得不那么严格，更难表述，因此请根据对您的应用程序最重要的内容进行判断。您还可以尝试通过使用硬性设备和模糊测试的健康组合来结合这两个领域的最佳效果。
备注
有关以下内容的更多信息 conftest.py's anatomy and pytest configuration can be found in the latter's documentation: conftest.py: local per-directory plugins 。
有了适当的基础工作，我们可以从一个简单的测试开始，该测试将尝试获取 /images 资源。将以下代码放入新的 tests/test_images.py 模块：
def test_list_images(client):
resp = client.simulate_get('/images')
assert resp.status_code == 200
assert resp.json == []
让我们试一试：：
$ pytest tests/test_images.py
========================= test session starts ==========================
platform linux -- Python 3.8.0, pytest-6.2.1, py-1.10.0, pluggy-0.13.1
rootdir: /falcon/tutorials/asgilook
collected 1 item
tests/test_images.py . [100%]
========================== 1 passed in 0.01s ===========================
成功！🎉
此时，我们的项目结构包含 asgilook 和 test 包，应如下所示：：
asgilook
├── .venv
├── asgilook
│ ├── __init__.py
│ ├── app.py
│ ├── asgi.py
│ ├── cache.py
│ ├── config.py
│ ├── images.py
│ └── store.py
└── tests
├── __init__.py
├── conftest.py
└── test_images.py
现在，我们需要更多的测试！尝试再添加几个测试用例到 tests/test_images.py ，使用 WSGI Testing Tutorial 作为您的指南(Falcon测试框架的接口对于ASGI和WSGI基本相同)。有关更多示例，请参阅 examples/asgilook/tests 在Falcon储存库里。
小技巧
对于更高级的测试用例，请参阅 falcon.testing.ASGIConductor 班级值得一看。
代码覆盖率¶
我们有多少 asgilook 代码是否包含在这些测试中？
获取覆盖范围报告的一种简单方法是使用 pytest-cov 插件(在PyPI上可用)。
安装后 pytest-cov 我们可以按如下方式生成覆盖报告：
$ pytest --cov=asgilook --cov-report=term-missing tests/
哦，哇！我们正好有全线覆盖，除了 asgilook/asgi.py 。如果需要，我们可以指示 coverage 要省略此模块，请将其列在 omit 的一节 .coveragerc 文件。
更重要的是，我们可以通过添加以下内容将当前的覆盖范围转变为需求 --cov-fail-under=100 (或任何其他百分比阈值)添加到我们的 pytest 指挥部。
备注
这个 pytest-cov 插件相当简单；更高级的测试策略，如混合不同类型的测试和/或在多个环境中运行相同的测试，很可能需要运行 coverage 直接，并结合结果。
现在怎么办？¶
恭喜您，您已经成功完成了FalconASGI教程！
不用说，我们的示例ASGI应用程序仍然可以在许多方面进行改进：
使映像存储持久化，并可跨工作进程重复使用。也许是通过使用数据库？
改进格式错误图像的错误处理。
检查Pillow如何以及何时释放GIL，并调优卸载到线程池执行器的内容。
测试 Pillow-SIMD 以提高性能。
通过以下方式发布图像上传事件
SSE或 WebSockets 。.还有更多(正如他们所说，欢迎补丁)！
与同步版本相比，异步代码有时更难设计和推理。如果您遇到任何问题，我们友好的社区可以回答您的问题并帮助您解决任何棘手的问题(另请参阅： Getting Help )。
