Sanic 下最好用 ORM 之 databases

发表于 更新于 分类于 阅读次数： 本文字数： 329 阅读时长 ≈ 1 分钟

前记

在使用 Sanic 的过程中对 ORM 的选择可谓是痛苦的，用过官方推荐的 Tortoise ORM，也用过 SQLAlchemy

比起使用 ORM 我更喜欢原生 SQL，当我看见 databases 的时候我发现它满足了的我所有要求，支持异步驱动 aiomysql，支持原生 SQL 写法，还封装进了 SQLAlchemy，只要你想你也可以把 databases 当作 SQLAlchemy 使用

安装

安装 databases

1
pip install databases

安装数据库驱动

1
pip install aiomysql

在安装 databases 的时候会自动的安装 SQLAlchemy 目前已经支持 1.4 版本

配置

利用监听器去控制数据库的连接和断开，并将句柄放入到应用上下文中 app.ctx.db

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
./server.py

from sanic import Sanic
from databases import Database

app = Sanic('jakehu')

# 数据库
database = Database('mysql://localhost/example')
# 服务开启之前
@app.listener("before_server_start")
async def startup_db(app, loop):
    await database.connect()
# 服务结束之后
@app.listener("after_server_stop")
async def shutdown_db(app, loop):
    await database.disconnect()
app.ctx.db = database

@app.get("/")
async def foo_handler(request):
    return text("Hello World!")

app.run(host='0.0.0.0', debug=True, auto_reload=True)

使用

最后看如何在函数中使用，利用应用上下文中的句柄进行操作，最后在利用_mapping 属性进行转换

1
2
3
4
5
6
7
@app.get("/")
async def foo_handler(request):
    database= request.app.ctx.db
    query = "SELECT * FROM table"
    rows = await database.fetch_all(query=query)
    result = [dict(row._mapping) for row in rows]
    return json(result)

关于如何将 sqlalchemy.engine.row.Row 转化为 dict 可以参考

示例