Sanic十六:Sanic + 异步orm之SQLAlchemy

 

Sanic是异步库,想要发挥其强大的性能,当需要使用第三方库的时候,就需要使用异步的库,在python中,异步orm较为常见的就两个可,一个SQLAlchemy,一个Tortoise-ORM

SQLAlchemy 在1.4版本之后,已经支持异步了,既然要用异步,那同步库的PyMYSQL肯就就不能满足了,所以需要用异步库aiomysql

 

SQLAlchemy官网:https://www.sqlalchemy.org/

SQLAlchemy中文网站:https://www.osgeo.cn/sqlalchemy/index.html

aiomysql官网:https://aiomysql.readthedocs.io/en/latest/

 

安装SQLAlchemy: pip install SQLAlchemy

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

安装aiomysql: pip install aiomysql

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

使用示例

先确保有数据库:库名随便取,比如test

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

创建模型

Sanic十六:Sanic + 异步orm之SQLAlchemy

由于没有sqlalchemy1.4以后的教程太少,没有找到用aiomysql作为驱动的映射教程,所以这里将对象映射到数据库的操作还是用mysql,系统运行中用aiomysql

Sanic十六:Sanic + 异步orm之SQLAlchemy

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

orm与app绑定,这里就用异步的aiomysql了

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

在中间件中定义获取数据库session和释放资源

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

路由

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

 插入数据

Sanic十六:Sanic + 异步orm之SQLAlchemy

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

查询

Sanic十六:Sanic + 异步orm之SQLAlchemy

 

models

from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class BaseModel(Base):
__abstract__ = True
id = Column(Integer, primary_key=True, comment='id,主键')


class User(BaseModel):
""" 用户表 """
__tablename__ = "user"
name = Column(String(64), comment='名字')
age = Column(Integer, comment='年龄')

def to_dict(self):
return {"id": self.id, "name": self.name, "age": self.age}

 

模型映射到数据库

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import Base

# 导入相应的模块
engine = create_engine("mysql+pymysql://root:123456@localhost/test")

# 创建session对象
session = sessionmaker(engine)()

# 创建表,执行所有BaseModel类的子类
Base.metadata.create_all(engine)

# 提交,必须
session.commit()

 

业务逻辑

from contextvars import ContextVar

from sanic import Sanic, response
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

from models import User

app = Sanic("SanicAndSqlalchemy")

# 创建异步数据库引擎
bind = create_async_engine("mysql+aiomysql://root:123456@localhost/test", echo=True)

_base_model_session_ctx = ContextVar("session")


@app.middleware("request")
async def inject_session(request):
"""
请求中间件
创建一个可用的 AsyncSession 对象并且将其绑定至 request.ctx 中,
而 _base_model_session_ctx 也会在这是被赋予可用的值,
如果需要在其他地方使用 session 对象(而非从 request.ctx 中取值),该全局变量或许能帮助您(它是线程安全的)。
"""
request.ctx.session = sessionmaker(bind, AsyncSession, expire_on_commit=False)()
request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)


@app.middleware("response")
async def close_session(request, response):
""" 响应中间件,将创建的 AsyncSession 关闭,并重置 _base_model_session_ctx 的值,进而释放资源 """
if hasattr(request.ctx, "session_ctx_token"):
_base_model_session_ctx.reset(request.ctx.session_ctx_token)
await request.ctx.session.close()


@app.post("/user")
async def create_user(request):
session = request.ctx.session
async with session.begin():
user = User(**request.json)
session.add(user)
return response.json(user.to_dict())


@app.get("/user/<user_id:int>")
async def get_user(request, user_id):
session = request.ctx.session
async with session.begin():
result = await session.execute(select(User).where(User.id == user_id))
user = result.scalar()
return response.json({'code': 200, 'message': '查询成功', 'data': user.to_dict() if user else {}})


if __name__ == '__main__':
import uvicorn

uvicorn.run('main:app', host='0.0.0.0', port=8000, debug=True)

 

上一篇:Mybatis


下一篇:django-orm数据迁移存在的问题