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可以参考

示例