fastapi 整合 sqlalchmey -续篇复盘：使用asynccontextmanager 上下文管理器的使用
1 前言
上一篇提到了一个关于协程异步使用中使用上下文的方式解决事务知道的处理的问题， 后来再回看一下其实，实现alchemical那种方式也很简单。而且只需要引入一个库就可以实现和同步方式一样的处理的方案。
sqlalchemy异步方式几个方案：
1：sqlalchemy 原生的提供的异步
sqlalchemy.ext.asyncio import AsyncSession
2：sqlalchemy +databases (官网案例)
以下的一些示例，主要重点关注的点是如何的使用异步上下文管理器，对我们的sessionmaker进行包装！
2 同步方式（再贴一次代码）
使用上下文的处理方式：
from sqlalchemy import Column, String, create_engine, Integer, Text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import time
import urllib
from sqlalchemy.orm import scoped_session
# - 1：引入数据库和会话引擎；
# 连接格式： 用户名:密码@localhost:端口/数据库名
password = 'xxxxxxxx'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@x.x.x.x/ces'
engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)
# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；
# expire_on_commit 是 现的一种方法是使用上下文管理器暂时expire_on_commit关闭
db_session = scoped_session(sessionmaker(bind=engine, expire_on_commit=True))
def GetSession():
return db_session()
from contextlib import contextmanager
@contextmanager
def session_scope():
session = GetSession()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()
# - 4：引入ORM模型的基础的基类declarative，所有模型基类都是它；
Base = declarative_base()
# 定义User对象:
class User(Base):
# 表的名字:
__tablename__ = 'users'
# 表的结构:
id = Column(Integer, autoincrement=True, primary_key=True, unique=True, nullable=False)
name = Column(String(50), nullable=False)
sex = Column(String(4), nullable=False)
code = Column(String(50), nullable=False)
create_time = Column(String(50), nullable=False)
updater = Column(String(32))
update_time = Column(String(20), nullable=False, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
onupdate=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
def createTable():
# 创建表
Base.metadata.create_all(engine)
def insertData():
# 插入操作
# - 5：使用的我们的会话实例对象进行具体的CURD操作。
with session_scope() as session:
# 创建新User对象:
local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
new_user = User(name='xiaoxao32', sex='23', code='43535334', create_time=local_time, update_time=local_time)
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
def selectData():
# 查询操作
# - 5：使用的我们的会话实例对象进行具体的CURD操作。
with session_scope() as session:
# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
user = session.query(User).filter(User.id == '2' and User.name == 'xiaoxao').one()
print('name:', user.name)
print('当前会话ID：', id(session))
# createTable()
def task1():
insertData()
def task2():
selectData()
#
from threading import Thread
for i in range(20):
t = Thread(target=task1)
t.start()
t = Thread(target=task2)
t.start()
3 异步协程方式
这个是本笔记中主要记录的一点！！！！
主要还是利用from contextlib import asynccontextmanager 的这种来处理！完整代码：
from typing import AsyncGenerator, Any
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import Column, String, create_engine, Integer, Text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import time
import urllib
from sqlalchemy.orm import scoped_session
from contextlib import asynccontextmanager
password = 'xxxxxxxxxxxxxxx'
url = f'postgresql+asyncpg://postgres:{urllib.parse.quote_plus(password)}@xx.x.x.x:5432/ces'
engine = create_async_engine(
url,
future=True,
echo=True,
)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
# - 4：引入ORM模型的基础的基类declarative，所有模型基类都是它；
Base = declarative_base()
class User(Base):
# 表的名字:
__tablename__ = 'users'
# 表的结构:
id = Column(Integer, autoincrement=True, primary_key=True, unique=True, nullable=False)
name = Column(String(50), nullable=False)
sex = Column(String(4), nullable=False)
code = Column(String(50), nullable=False)
create_time = Column(String(50), nullable=False)
updater = Column(String(32))
update_time = Column(String(20), nullable=False, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
onupdate=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
# 需要使用这个来装饰一下，才可以使用with
@asynccontextmanager
async def get_db() -> AsyncGenerator:
session = async_session()
try:
yield session
await session.commit()
except SQLAlchemyError as ex:
await session.rollback()
raise ex
finally:
await session.close()
async def insertData():
# 插入操作
# - 5：使用的我们的会话实例对象进行具体的CURD操作。
async with get_db() as session:
# 创建新User对象:
local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
new_user = User(name='xiaoxao32', sex='23', code='43535334', create_time=local_time, update_time=local_time)
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
from sqlalchemy import Column, String, select
async def selectData():
# 查询操作
# - 5：使用的我们的会话实例对象进行具体的CURD操作。
async with get_db() as session:
# 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
stmt = select(User).where(User.name == 'xiaoxao').where(User.id == 10)
result = await session.execute(stmt)
instance = result.scalars().first()
print(' 输出结果:', instance.name)
print(' 输出结果:', instance.code)
import asyncio
asyncio.run(selectData())
比较关键的地方主要是
# 需要使用这个来装饰一下，才可以使用with
@asynccontextmanager
async def get_db() -> AsyncGenerator:
session = async_session()
try:
yield session
await session.commit()
except SQLAlchemyError as ex:
await session.rollback()
raise ex
finally:
await session.close()
有了它之后，我们就可以在随意的地方，直接的使用上下文方式来提交处理数据操作逻辑了~
PS:不过需要注意的一点就是，使用协程异步的方式之后，相关的查询的有些不一样了！和同步使用的查询有些区别！
更多的使用方式建议查看官网的文档--推荐查看中文文档：https://www.osgeo.cn/sqlalchemy/tutorial
4：sqlalchemy官网文档更简单一种方式
直接的使用engine的引擎的事务来处理，且支持上下文的方式！
examples.asyncio.async_orm 源代码
"""Illustrates use of the sqlalchemy.ext.asyncio.AsyncSession object
for asynchronous ORM use.
"""
import asyncio
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.future import select
from sqlalchemy.orm import relationship
from sqlalchemy.orm import selectinload
Base = declarative_base()
class A(Base):
__tablename__ = "a"
id = Column(Integer, primary_key=True)
data = Column(String)
bs = relationship("B")
class B(Base):
__tablename__ = "b"
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey("a.id"))
data = Column(String)
async def async_main():
"""Main program function."""
engine = create_async_engine(
"postgresql+asyncpg://scott:tiger@localhost/test",
echo=True,
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
async with session.begin():
session.add_all(
[
A(bs=[B(), B()], data="a1"),
A(bs=[B()], data="a2"),
A(bs=[B(), B()], data="a3"),
]
)
# for relationship loading, eager loading should be applied.
stmt = select(A).options(selectinload(A.bs))
# AsyncSession.execute() is used for 2.0 style ORM execution
# (same as the synchronous API).
result = await session.execute(stmt)
# result is a buffered Result object.
for a1 in result.scalars():
print(a1)
for b1 in a1.bs:
print(b1)
# for streaming ORM results, AsyncSession.stream() may be used.
result = await session.stream(stmt)
# result is a streaming AsyncResult object.
async for a1 in result.scalars():
print(a1)
for b1 in a1.bs:
print(b1)
result = await session.execute(select(A).order_by(A.id))
a1 = result.scalars().first()
a1.data = "new data"
await session.commit()
asyncio.run(async_main())
5：fastapi官网提供的方案
主页是结合的databases和sqlalchemy来做异步处理。但是这种方式对应的sqlalchemy的模型就比较受限！主要是依赖：sqlalchemy.Table 来创建对应的模型。
from typing import List
import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel
# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
notes = sqlalchemy.Table(
"notes",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("text", sqlalchemy.String),
sqlalchemy.Column("completed", sqlalchemy.Boolean),
)
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)
class NoteIn(BaseModel):
text: str
completed: bool
class Note(BaseModel):
id: int
text: str
completed: bool
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
@app.get("/notes/", response_model=List[Note])
async def read_notes():
query = notes.select()
return await database.fetch_all(query)
@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
query = notes.insert().values(text=note.text, completed=note.completed)
last_record_id = await database.execute(query)
return {**note.dict(), "id": last_record_id}
6 单独使用database异步提供的方案
这个单独的使用database的话，主要是对原生的SQL语句的执行。示例如下：
# Create a database instance, and connect to it.
from databases import Database
database = Database('sqlite:///example.db')
await database.connect()
# Create a table.
query = """CREATE TABLE HighScores (id INTEGER PRIMARY KEY, name VARCHAR(100), score INTEGER)"""
await database.execute(query=query)
# Insert some data.
query = "INSERT INTO HighScores(name, score) VALUES (:name, :score)"
values = [
{"name": "Daisy", "score": 92},
{"name": "Neil", "score": 87},
{"name": "Carol", "score": 43},
]
await database.execute_many(query=query, values=values)
# Run a database query.
query = "SELECT * FROM HighScores"
rows = await database.fetch_all(query=query)
print('High Scores:', rows)
结尾
END
简书：https://www.jianshu.com/u/d6960089b087
掘金：https://juejin.cn/user/2963939079225608
公众号：微信搜【小儿来一壶枸杞酒泡茶】
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。 2：关于文章内容，部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。
原文作者：程序员小钟同学；如有侵权，请联系管理员删除。
请登录后发表评论
登录
全部评论
最新文章
- 深入FastAPI异步编程-之并发编程（15）-定义全局的支持全局上下文变量和局部上下文的对象可以任意传递了！
- 深入FastAPI异步编程-之并发编程（13）-纠错-修改基于aiojobs库结合aiojobs的和FastAPI框架的示例
- 深入FastAPI异步编程-之并发编程（13）-基于aiojobs库实现类似Background Tasks的效果
- 深入FastAPI异步编程-之并发编程（12）-基于pyee异步事件驱动通知
- 深入FastAPI异步编程-之并发编程（11）-异步中的?同步原语
- 深入FastAPI异步编程-之并发编程（10）-?asyncio.wait 更精细化并发补充说明
- 深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明
- 深入FastAPI异步编程-之并发编程（8）-asyncio.gather的补充说明
- 深入FastAPI异步编程-之并发编程（7）番外篇~Windows系统下事件循环插曲
- 深入FastAPI异步编程-之并发编程（6）~事件循环
