ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > fastapi 整合 sqlalchmey ~干掉依赖注入实现会话创建的方式的几种！

fastapi 整合 sqlalchmey ~干掉依赖注入实现会话创建的方式的几种！

原创 数据库开发技术 作者：程序员小钟同学 时间：2021-10-23 18:43:05 0 删除 编辑

1 说明

之前线上业务都是基于PEEWEE这个ORM的来做的，部分的复杂业务逻辑直接则是使用的原生SQL。基于最近的最近刚好想把线上一个后台系统整合改进，替换一个新的orm,也为了当做学习，所以切换当下比较流程的orm框架Sqlalchmey。

目前Sqlalchmey最新的版本已经是1.4.26版本了，而且也对异步协程已经早支持了！所以必须得学习一波了！

当然其实对于ORM使用的争议问题，我这里不细说，其实ORM使用与否主要看你个人的业务和使用需求和场景而定，有些人偏向原生SQL，感觉那带劲！有些人可能觉得orm直观，所以各有所爱吧！

2 常用的SQL（结构化查询语言）语句

  • 数据定义语言【DDL】：主要处理表结构的创建和删除
  • 数据控制语言【DCL】：定义数据的访问和修改权限
  • 数据查询语言【DQL】：对数据的查询
  • 数据操作语言【DML】：对数据的增删改
  • 数据事务语言【TPL】：保持数据的一致性
  • 指针控制语言【CCL】：定义的是对查询的数据进行相关的逐行控制等

3 sqlalchmey初识

ORM提供的是SQL工具包和对象关系的映射一种方式。

3.1 sqlalchmey 同步方式下-基本使用流程

流程文字说明

  • 1：引入数据库和会话引擎；

    PS：engine = create_engine

  • 2：使用会话引擎对象绑定会话工程创建类；

    PS:sessionmaker(bind=engine)

  • 3：通过会话工程类创建具体的会话实例；

    PS:db_session = sessionmaker(bind=engine)

  • 4：引入ORM模型的基础的基类declarative，所有模型基类都是它；

    PS:Base = declarative_base()

  • 5：使用的我们的会话实例对象进行具体的CURD操作。

    PS: session = db_session()  # 创建会话

create_engine中的一些参数 

create_engine中的一些参数：
- pool_size=5,  # 连接池大小
- pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
- pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
- echo= True 输出执行SQL日志
- max_overflow=10, 超出 pool_size 后可允许的最大连接数，默认为 10, 这 10 个连接在使用过后，不放在 pool 中，而是被真正关闭的
- pool_timeout=30, 获取连接的超时阈值，默认为 30 秒
- poolclass=NullPool 表示禁用连接池的方式

3.2 sqlalchmey 流程示例说明

首先我这里使用的是最新版本的sqlalchmey，连接的postgresql数据看看，其实ORM已经屏蔽了底层连接处理，支持不同的数据库引擎。所以这个基本是保持一致的。只是引入的和需要安全的引擎不一样。

这里示例使用的还是同步的方式，协程的方式，后续在整合到Fastapi的时候，再展开！

示例参考来源：https://blog.csdn.net/u013992330/article/details/113655840

示例代码：

#!/usr/bin/evn python
# coding=utf-8

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

# - 1：引入数据库和会话引擎；
# 连接格式： 用户名:密码@localhost:端口/数据库名
password = 'xxxxxxxxxxxxx'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@127.0.0.1:5432/ces'
engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)

# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；
db_session = sessionmaker(bind=engine)

# - 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操作。
    session = db_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)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    session.close()


def selectData():
    # 查询操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_session()
    # 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
    user = session.query(User).filter(User.id == '1' and User.name == 'hsh4').one()
    print('name:', user.name)
    session.close()  # 关闭Session


def updateData():
    # 更新操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_session()
    users = session.query(User).filter_by(name="xiaoxiao").first()  # 查询条件
    users.code = "1000010"  # 更新操作
    session.add(users)  # 添加到会话
    session.commit()  # 提交即保存到数据库
    session.close()  # 关闭会话


def deleteData():
    # 删除操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_session()  # 创建会话
    delete_users = session.query(User).filter(User.id == "1").first()
    if delete_users:
        session.delete(delete_users)
        session.commit()
    session.close()  # 关闭会话


def dropTable():
    # 删除表
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_session()  # 创建会话
    session.execute('drop table users')
    session.commit()
    session.close()


createTable()
insertData()
selectData()
updateData()
deleteData()
# dropTable()

最新一些相关的操作：

3.3 sqlalchmey 全局会话对象

上面的示例中是在同一个线程的之中处理，完成，而且我们的每次的执行的相关的数据的提交的时候，都是创新新的会话对象来处理我们的数据库的连接的。

那如果我们的只是创建一个会话ID可以执行一连串的操作吗？


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

# - 1：引入数据库和会话引擎；
# 连接格式： 用户名:密码@localhost:端口/数据库名
password = 'xxxxxxxxx'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@xxx.xxx.x.xx:5432/ces'
engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)

# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；
db_session = sessionmaker(bind=engine)

# - 4：引入ORM模型的基础的基类declarative，所有模型基类都是它；
Base = declarative_base()

session = db_session()  # 创建一个全局的会话

# 定义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操作。
    # session = db_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)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    print("关闭了会话！！！")
    print('当前会话ID：', id(session))
    session.close()


def selectData():
    # 查询操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_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))
    session.close()  # 关闭Session


def updateData():
    # 更新操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_session()
    users = session.query(User).filter_by(name="xiaoxao32").first()  # 查询条件
    users.code = "1000010"  # 更新操作
    print('当前会话ID：',id(session))
    session.add(users)  # 添加到会话
    session.commit()  # 提交即保存到数据库
    session.close()  # 关闭会话


def deleteData():
    # 删除操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_session()  # 创建会话
    delete_users = session.query(User).filter(User.id == "1").first()
    if delete_users:
        session.delete(delete_users)
        session.commit()
    session.close()  # 关闭会话


def dropTable():
    # 删除表
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_session()  # 创建会话
    session.execute('drop table users')
    session.commit()
    session.close()



createTable()
insertData()
selectData()
updateData()
# deleteData()
# dropTable()

上面的示例和前面的最大的不同是：

且我们的执行相关的操作的时候，不再新家一个会话ID去处理。

最终发现也可以执行相关的操作，好奇，为什么我们的意见对会话执行关闭了，它还可以继续执行操作的呢？

查看关于close的介绍说明：

意思是在1.4版本中是可以自己重新创建额的一个新的出来，那我就好奇，重新创建的新的话应该是新的ID，这个就不思其解了！有知道的话，希望能对小弟我指点一下。

3.4 sqlalchmey 多线程机制会话对象探究

上面的示例中【使用全局会话ID】是在同一个线程的之中处理完成。但是这种全局的会话ID在多线程情况下是否是线程安全的呢？

PS:关于session = db_session()  # 创建会话

当执行会话创建的时候，其实这个不会马上从eengine连接池里面获取一个连接，而是当执行到我们的对数据具体的一些操作的时候才会执行，直到我们的对相关数据的操作整个整过过程也会一直持有，当数据操作完成后才关闭，才回去释放。

1：需要注意的一点是：会话需要执行close关闭，这个连接的才会重新被回收

测试验证一下在多线的环境环境的使用全局会话ID下问题：


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 = 'sssaaaaaaaxxxxxx'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@xxxxxxxx:5432/ces'
engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)

# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；
db_session = sessionmaker(bind=engine)
# Session = scoped_session(db_session)
# 将连接交还给连接池

# session.remove(
# Session.remove()
# - 4：引入ORM模型的基础的基类declarative，所有模型基类都是它；
Base = declarative_base()

session = db_session()  # 创建会话

# 定义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操作。
    # session = db_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)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    print("关闭了会话！！！")
    print('当前会话ID：', id(session))
    session.close()



def selectData():
    # 查询操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_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))
    session.close()  # 关闭Session


def updateData():
    # 更新操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_session()
    users = session.query(User).filter_by(name="xiaoxao32").first()  # 查询条件
    users.code = "1000010"  # 更新操作
    print('当前会话ID：',id(session))
    session.add(users)  # 添加到会话
    session.commit()  # 提交即保存到数据库
    session.close()  # 关闭会话


def deleteData():
    # 删除操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_session()  # 创建会话
    delete_users = session.query(User).filter(User.id == "1").first()
    if delete_users:
        session.delete(delete_users)
        session.commit()
    session.close()  # 关闭会话


def dropTable():
    # 删除表
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_session()  # 创建会话
    session.execute('drop table users')
    session.commit()
    session.close()



# 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()

# insertData()
# selectData()
# updateData()
# deleteData()
# dropTable()

只是上面的代码你会发现出现了异常：

说明了我们的全局的会话对象在多线程环境是不安全滴。

(1)多线程安全的解决方案1：--基于scoped_session

scoped_session内部其实是基于threding.local的方式来解决线程安全的问题。

而且当使用基于scoped_session的之后，我们的会话对象也还可以是一个全局的会话对象，只是返回的额是一个scoped_session，并且这个的会话的回收和挂你，不再是调用close，而只是接的使用remove的方法就可以归还回收连接的对象。

所以最终的方案的代码如下：


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 = 'xxxxxxxxxx'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@4xxxxxxxxxx.x.x.x:5432/ces'
engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)

# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；


sessionmakerFac = sessionmaker(bind=engine)
# ==========================================
session = scoped_session(sessionmakerFac) # 创建会话
# 将连接交还给连接池
# session.remove(）
# - 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 insertData():
    # 插入操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_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)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    print("关闭了会话！！！")
    print('当前会话ID：', id(session))
    # 将连接交还给连接池
    session.remove()


def selectData():
    # 查询操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    # session = db_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))
    # session.close()  # 关闭Session
    # 将连接交还给连接池
    session.remove()



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()


在上面的示例中我们的是直接的使用了scoped_session返回的Session对象来操作数据了。为啥不继续使用scoped_session返回的Session对象再来创建一个新的Session呢？

session = scoped_session(sessionmakerFac) 

session_01 = session()
session_02 = session()
print(id(session_01))
print(id(session_02))

从上面看的话：scoped_session其实是一个单例模式。初始化的对象只有一个。

(2)多线程安全的解决方案2：--创建不同的会话对象ID处理 


省略部分的代码·····

engine = create_engine(url=uri, echo_pool=True, echo=False, pool_size=10)

# - 2：使用会话引擎对象绑定会话工程创建类-sessionmaker(bind=engine):
# - 3：通过会话工程类创建具体的会话实例；

db_session = sessionmaker(bind=engine)


# - 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操作。
    session = db_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)
    # 提交即保存到数据库:
    session.commit()
    # 关闭session:
    print("关闭了会话！！！")
    print('当前会话ID：', id(session))
    session.close()



def selectData():
    # 查询操作
    # - 5：使用的我们的会话实例对象进行具体的CURD操作。
    session = db_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))
    session.close()  # 关闭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()

4 sqlalchmey使用contextmanager封装使用会话：

封装的主要目的是，为了能后续的操作数据的时候，相关的相关的事务的处理。

参考我以前的使用peewee的使用的习惯的话，我们会使用上下的的方式简单等着一个使用：

  • PooledPostgresqlExtDatabase的代码示例：
database = PooledPostgresqlExtDatabase(
    db_config_const.get('db_name'),
    max_connections=db_config_const.get('max_connections'),
    stale_timeout=db_config_const.get('stale_timeout'),  # 5 minutes.
    timeout=db_config_const.get('timeout'),
    **{'user': db_config_const.get('db_user'), 'host': db_config_const.get('db_host'),
       'password': db_config_const.get('db_pass'), 'port': db_config_const.get('db_port')})

def GetSession():
    return database

def GetSessionAtomic():
    return database.atomic()

def session_scope_atomic():
    return GetSessionAtomic()
    # if session.is_closed():
    #     # session.connection_context()
    #     session.connect()



from contextlib import contextmanager


@contextmanager
def session_scope():
    session = GetSession()
    if session.is_closed():
        # session.connection_context()
        session.connect()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def session_scope2():
    session = GetSession()
    with session.manual_commit():
        session.begin()  # Begin transaction explicitly.
        try:
            yield session
        except:
            session.rollback()  # Rollback -- an error occurred.
            raise
        else:
            try:
                session.commit()  # Attempt to commit changes.
            except:
                session.rollback()  # Error committing, rollback.
                raise

改到我们的sqlalchmey其实方式流程也是一样的：

  1. 建立数据库会话
  2. 随时自动的创建数据库事务
  3. 对执行数据操作异常的捕获，回滚事务
  4. 当执行数据操作完成，归还连接，到连接池。
  5. 在执行的关闭的前，自动的进行相关的的commit提交

这种的封装的方式，可以让我们的数据库的操作完全的独立于一些请求的上下文的处理， 特别是是使用fastapi的时候，我个人不喜欢使用依赖注入的方式来获取连接的会话的对象。所以我会使用下面的这种的方式.

完整的示例如下：

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 = 'xiao@#SsLIV(&%$cPha*m)+=asoEW-#'
uri = f'postgresql+psycopg2://postgres:{urllib.parse.quote_plus(password)}@47.99.189.42:5432/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()

总结：通过上面的这种方式，我们的后续在同步的一些框架的使用的话，直接的使用的这种上下扥操作的方式就可以对我们的数据库的进行的相关的的操作了！

不需要通过使用的吗依赖注入的方式来实现的获取会话对象了，且上面的这种方式是线程的安全的，在多线程的情况依然可以正常的使用。

5 sqlalchmey 一些查询条件的使用

5.1 等值过滤器查询 

session.query.filter(User.name == 'xiaozhong')
session.query.filter(User.age == 31)

5.2 不等于（!=,<,>,<=,>=） 

session.query.filter(User.name != 'xiaozhong')
session.query.filter(User.age > 31)

5.3 模糊匹配like： 

#大小写敏感
session.query.filter(User.name.like('%xiao%'))
# 大小写不敏感
session.query.filter(User.name.ilike('%xiao%'))

5.4 in过滤 

表示查询在里面的
session.query.filter(~User.name.in_(['xiao', 'zhong', 'niihao']))
in子表达式：
~ 表示的是在不里面：
session.query.filter(~User.name.in_(session.query.filter(~User.name.in_(['xiao', 'zhong', 'niihao'])))

5.5 空值和非空值 

session.query.filter(User.name == None)
session.query.filter(User.name.is_(None))
session.query.filter(User.name != None)
session.query.filter(User.name.isnot(None))

5.6 and_和 or_ 

from sqlalchemy import and_
方案1：
session.query.filter(and_(User.name == 'xsasda', User.age == 31))
方案2：
session.query.filter(User.name == 'xsasda', User.age == 31)
方案1：
session.query.query.filter(User.name == 'xsasda').filter(User.age == 31)
====
from sqlalchemy import or_
query.filter(or_(User.name == 'xsasda', User.name == 'xsasda2'))

5.7 join 

query.join(Address, User.id==Address.user_id)

# 其他sql示例
result = session.query(User.username,func.count(Article.id)).join(Article,User.id==Article.uid).\
    group_by(User.id).order_by(func.count(Article.id).desc()).all()

5.8 窗口函数 

result = session.query(User.username,func.count(Article.id)).join(Article,User.id==Article.uid).\
    group_by(User.id).order_by(func.count(Article.id).desc()).all()

更多的学习可以参考：https://www.cnblogs.com/zhongyehai/p/11828337.html

6 sqlalchmey结合fastapi中的使用（异步协程示例）

上面的示例中，其实已经算是我们的对会话的管理的一种封装和引入，如果我们的在fastapi中使用的是多线程的模式的话，那么我的可以直接的使用上面的那种的封装的方式来处理数据库的操作就可以了！

那这里我再介绍一下其他的处理的方式，这里主要是列举当前多数的人的使用的方式：

  • 1：在依赖注入的实现对我们的会话对的创建（我个人非常不喜欢）
  • 2: 使用中间件的方式来创建的我们的会话ID（似乎有点意思，而且，整的会话可以在同一个线程的内的进行重复的使用）-fastapi_sqlalchemy 的设计

fastapi_sqlalchemy 具体地址为：https://hub.fastgit.org/mfreeborn/fastapi-sqlalchemy

  • 3：这种就是我的上面那只使用上下文的方式来处理了(不过不建议使用在于协程路由内)

  • 4：使用Alchemical，可以实现我的上述类似3 的处理方式

    Alchemical具体地址为：https://hub.fastgit.org/miguelgrinberg/alchemical

6.1 依赖注入法（1）

主要流程思路是：

1：首先还是创建的全局的会话async_session

DATABASE_URL = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_SERVER}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}"

engine = create_async_engine(DATABASE_URL, future=True, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

2：然后就是定义依赖注入方法

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

3：定义相关的数据表CRUD模型的操作，如get_user_by_username：

from db.models.user import User
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select


async def create_user(
    db_session: AsyncSession, username: str, hased_password: str, avatar: str
):
    new_user = User(username=username, hashed_password=hased_password, avatar=avatar)
    db_session.add(new_user)
    await db_session.commit()
    await db_session.flush()
    return new_user


async def get_user_by_username(db_session: AsyncSession, username: str) -> User:
    result = await db_session.execute(select(User).where(User.username == username))
    return result.scalars().first()

4：然后在具体的路由进行依赖注入的实例化

@router.post("/login")
async def login_for_access_token(async_session=Depends(get_session)
):
    user = await 如get_user_by_username(
        async_session, form_data.username
    )
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "user_id": user.id}

6.2 依赖注入法（2）

案例来源：

https://hub.fastgit.org/grillazz/fastapi-sqlalchemy-asyncpg

1：首先还是创建的全局的会话async_session 和依赖方法

from typing import AsyncGenerator

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

global_settings = config.get_settings()
url = global_settings.asyncpg_url

engine = create_async_engine(
    url,
    future=True,
    echo=True,
)

# expire_on_commit=False will prevent attributes from being expired after commit.

async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


# Dependency
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()

2：定义模型的Base的基础

from typing import Any

from fastapi import HTTPException, status
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.declarative import as_declarative, declared_attr


@as_declarative()
class Base:
    id: Any
    __name__: str
    # Generate __tablename__ automatically

    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

    async def save(self, db_session: AsyncSession):
        """
        :param db_session:
        :return:
        """
        try:
            db_session.add(self)
            return await db_session.commit()
        except SQLAlchemyError as ex:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=repr(ex))

    async def delete(self, db_session: AsyncSession):
        """
        :param db_session:
        :return:
        """
        try:
            await db_session.delete(self)
            await db_session.commit()
            return True
        except SQLAlchemyError as ex:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=repr(ex))

    async def update(self, db_session: AsyncSession, **kwargs):
        """
        :param db_session:
        :param kwargs:
        :return:
        """
        for k, v in kwargs.items():
            setattr(self, k, v)
        await self.save(db_session)

3：然后是具体的模型的实例的实现：

  
import uuid

from fastapi import HTTPException, status
from sqlalchemy import Column, String, select
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.asyncio import AsyncSession

from the_app.models.base import Base


class Nonsense(Base):
    __tablename__ = "nonsense"
    id = Column(UUID(as_uuid=True), unique=True, default=uuid.uuid4, autoincrement=True)
    name = Column(String, nullable=False, primary_key=True, unique=True)
    description = Column(String, nullable=False)

    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    @classmethod
    async def find(cls, db_session: AsyncSession, name: str):
        """
        :param db_session:
        :param name:
        :return:
        """
        stmt = select(cls).where(cls.name == name)
        result = await db_session.execute(stmt)
        instance = result.scalars().first()
        if instance is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail={"Record not found": f"There is no record for requested name value : {name}"},
            )
        else:
            return instance

4：对数据异步的操作

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

from the_app.database import get_db
from the_app.models.nonsense import Nonsense
from the_app.schemas.nnonsense import NonsenseResponse, NonsenseSchema

router = APIRouter(prefix="/v1/nonsense")


@router.post("/", status_code=status.HTTP_201_CREATED, response_model=NonsenseResponse)
async def create_nonsense(payload: NonsenseSchema, db_session: AsyncSession = Depends(get_db)):
    nonsense = Nonsense(**payload.dict())
    await nonsense.save(db_session)
    return nonsense


@router.get("/", response_model=NonsenseResponse)
async def find_nonsense(
    name: str,
    db_session: AsyncSession = Depends(get_db),
):
    return await Nonsense.find(db_session, name)


@router.delete("/")
async def delete_nonsense(name: str, db_session: AsyncSession = Depends(get_db)):
    nonsense = await Nonsense.find(db_session, name)
    return await nonsense.delete(nonsense, db_session)


@router.patch("/", response_model=NonsenseResponse)
async def update_nonsense(
    payload: NonsenseSchema,
    name: str,
    db_session: AsyncSession = Depends(get_db),
):
    nonsense = await Nonsense.find(db_session, name)
    await nonsense.update(db_session, **payload.dict())
    return nonsense

6.3 使用的中间件的方式：

1:使中间的方式的，它实现，感觉好像应该是存在问题，因为整个过程我看到是如果我们的数据库是查询耗时的，我它不是使用的协程的尝试处理的，这样估计可能会有问题，不过暂时还没来得及进行测试验证。

from contextvars import ContextVar
from typing import Dict, Optional, Union

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import Session, sessionmaker
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.types import ASGIApp

from fastapi_sqlalchemy.exceptions import MissingSessionError, SessionNotInitialisedError

_Session: sessionmaker = None
_session: ContextVar[Optional[Session]] = ContextVar("_session", default=None)


class DBSessionMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: ASGIApp,
        db_url: Optional[Union[str, URL]] = None,
        custom_engine: Optional[Engine] = None,
        engine_args: Dict = None,
        session_args: Dict = None,
        commit_on_exit: bool = False,
    ):
        super().__init__(app)
        global _Session
        engine_args = engine_args or {}
        self.commit_on_exit = commit_on_exit

        session_args = session_args or {}
        if not custom_engine and not db_url:
            raise ValueError("You need to pass a db_url or a custom_engine parameter.")
        if not custom_engine:
            engine = create_engine(db_url, **engine_args)
        else:
            engine = custom_engine
        _Session = sessionmaker(bind=engine, **session_args)

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        with db(commit_on_exit=self.commit_on_exit):
            response = await call_next(request)
        return response


class DBSessionMeta(type):
    # using this metaclass means that we can access db.session as a property at a class level,
    # rather than db().session
    @property
    def session(self) -> Session:
        """Return an instance of Session local to the current async context."""
        if _Session is None:
            raise SessionNotInitialisedError

        session = _session.get()
        if session is None:
            raise MissingSessionError

        return session


class DBSession(metaclass=DBSessionMeta):
    def __init__(self, session_args: Dict = None, commit_on_exit: bool = False):
        self.token = None
        self.session_args = session_args or {}
        self.commit_on_exit = commit_on_exit

    def __enter__(self):
        if not isinstance(_Session, sessionmaker):
            raise SessionNotInitialisedError
        self.token = _session.set(_Session(**self.session_args))
        return type(self)

    def __exit__(self, exc_type, exc_value, traceback):
        sess = _session.get()
        if exc_type is not None:
            sess.rollback()

        if self.commit_on_exit:
            sess.commit()

        sess.close()
        _session.reset(self.token)


db: DBSessionMeta = DBSession

这个上面的看得出，它是定义个全局的db: DBSessionMeta = DBSession，里面而已是使用  global _Session的方法，那加入我的多线程的情况使用的话，估计这个····

模型相关：

from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, index=True)
    credit_balance = Column(Integer)
    username = Column(String)
    email = Column(String)
    password = Column(String)
    time_created = Column(DateTime(timezone=True), server_default=func.now())
    time_updated = Column(DateTime(timezone=True), onupdate=func.now())

    

fastapi_sqlalchemy 中的db使用方式：


import os
import sys
from datetime import timedelta

from dotenv import load_dotenv
from fastapi import File, UploadFile, Depends, FastAPI, HTTPException, status
from fastapi_sqlalchemy import DBSessionMiddleware, db
from logzero import logger
from models import User as ModelUser
load_dotenv('.env')

app = FastAPI()
app.add_middleware(DBSessionMiddleware, db_url=os.environ['DATABASE_URL'])

@app.post('/get_account_info')
async def login_for_account_info():
    user = db.session.query(ModelUser).filter(ModelUser.id == user_id).first()

    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Incorrect username or password',
            headers={'WWW-Authenticate': 'Bearer'},
        )

    account_info = {
        'id': user.id,
        'credit_balance': user.credit_balance,
        'username': user.username,
        'email': user.email,
        'time_created': user.time_created,
        'time_updated': user.time_updated
    }

    return {'account_info': account_info}

6.4 Alchemical的上下文的方式，且支持协程的 

import asyncio
import sys
from fastapi import FastAPI
from sqlalchemy import Column, Integer, String
from alchemical.aio import Alchemical

app = FastAPI()
db = Alchemical('sqlite:///app.db')


class User(db.Model):
    id = Column(Integer, primary_key=True)
    name = Column(String(128))


@app.get('/')
async def index():
    async with db.Session() as session:
        users = await session.scalars(User.select())
        return {'users': [u.name for u in users]}


async def add():
    """Add test user."""
    async with db.begin() as session:
        session.add(User(name='test'))


if sys.argv[1] == 'init':
    asyncio.run(db.create_all())
elif sys.argv[1] == 'add':
    asyncio.run(add())
else:
    raise RuntimeError('Invalid command')

以上是我的认知的几种的总结。

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【小儿来一壶枸杞酒泡茶】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822

  • 1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。
  • 2：关于文章内容，部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： fastapi 整合 sqlalchmey -续篇复盘：使用asynccontextmanager 上下文管理器的使用
下一篇： Fastapi框架 【开源（自用）脚手架】
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111698

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593