Quart/peewee async 踩坑记录

Table of Contents

这破博客已经一个月没有更新，感觉快要凉了。然而本次更的也并不是技术性文章，仅是前几天 HackDay(24H猝死日)上踩的一些坑。事实上也不算什么坑，毕竟我才只写了 700 来行代码

技术栈为

Quart 0.5.0
peewee-async 0.5.12

首先说一下为什么使用这两个的组合

Quart 是一个拥有和 Flask 一样 API 的基于 asyncio 的 Web 框架

peewee-async 则是不得不用，后面会提到

下面开始聊一下使用体验

extension
#

它的 tutorial 中有提到使用 Flask 的扩展 flask-login。但貌似它也并没有说它目前可以完美支持所有的 Flask 扩展，比如我想使用的 flask_cors 就不行。目前已知支持的扩展可参考这个列表 supported-extensions

迷之 Bug
#

此 Bug 我现在已经不能复现，我也不知道当时改过什么，全程梦游

这里在上传文件时抛出了一个 KeyError: 0

TraceBack
#

这个是有记录的，我的脚本名字是 analysis.py，从数据库中取指定 interview_id 的记录

interview = await manager.get(
    Interview,
    Interview.interview_id == interview_id
)

报错信息如下

peewee-async 的代码里是这么写的

try:
    result = yield from self.execute(query)
    return list(result)[0]
except IndexError:
    raise model.DoesNotExist

第一个坑爹的地方，这个 model.DoesNotExist 不是说你这个 model 不存在，而是指数据库中没有符合查询条件的结果

第二个坑爹的地方，这个 traceback 的输出貌似是有问题的。据我的理解异常应该是

  File: quart.py line xx
  File: quart.py line xx
  File: peewee_async.py line xx
    return list(result)[0]
IndexError: index out of range

During handling of the above exception, another exception occurred:
  File: peewee_async.py line xx
    raise model.DoesNotExist

WSGI
#

基于 asyncio 的 Web 框架没有一个支持 WSGI 的，这是理所当然的。Quart 自己提供了一个 Gunicorn 的 worker quart.worker.GunicornUVLoopWorker，这点还是比较方便的。另外 Quart 支持 ASGI

await
#

哪个 api 需要 await
我该 await 谁

其实这点还是很好理解的，典型的两个错误如下

# return render_template('xx.html')
return await render_template('xx.html')

# await request.files['filename']
(await request.files)['filename']

ORM
#

比较成熟的 async 的 ORM 目前还是比较少的，或者说能用的仅有下面两个

GINO，SQLAlchemy 风格，不过目前仅支持 PostgreSQL
peewee-async，peewee 风格，支持 PostgreSQL 和 MySQL

GINO 文档中说它可以进行事务，peewee_async 则说目前事务功能在测试中。所以如果想上生产环境，还是需要严格的测试一下

其实这里不用 async 的 ORM，我觉的也是可以的。这种模式被称为 partial async。或者直接使用 ThreadExecutor

真的是 Flask 么
#

Sanic 也是一个 Flask-like Web 框架，可以它没有全局的 request。Quart 目前看起来是最像 Flask 的框架了

@app.route('/')
async def hello():
    return iter([b'hello', b'world'])

这个在 Flask 上是错误的(我不是说 async)，但在 Quart 中能正常运行。可以参考 response.py#L89

self.response: AsyncIterable[bytes]
if isinstance(response, (str, bytes)):
    self.set_data(response)  # type: ignore
else:
    self.response = _ensure_aiter(response)  # type: ignore
    self.push_promises: Set[str] = set()

def set_data(self, data: AnyStr) -> None:
    if isinstance(data, str):
        bytes_data = data.encode(self.charset)
    else:
        bytes_data = data
    self.response = _ensure_aiter([bytes_data])

def _ensure_aiter(
        iter_: Union[AsyncGenerator[bytes, None], Iterable],
) -> AsyncGenerator[bytes, None]:
    if isasyncgen(iter_):
        return iter_  # type: ignore
    else:
        async def aiter() -> AsyncGenerator[bytes, None]:
            for data in iter_:  # type: ignore
                yield data
    return aiter()

这样做的好处就是可以进行流式处理，然后我就顺手写了这么一段智障代码

@app.route('/')
async def hello():
    with open(__file__, 'rb') as f:
        return iter(f.readline, b'')

真是糟糕的一天，感谢开源社区为我提供了这么多的欢乐

extension#

迷之 Bug#

TraceBack#

WSGI#

await#

ORM#

真的是 Flask 么#