这破博客已经一个月没有更新,感觉快要凉了。然而本次更的也并不是技术性文章,仅是前几天 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'')
真是糟糕的一天,感谢开源社区为我提供了这么多的欢乐