SQLAlchemy - 模块文件以及增删改查(CURD操作)

创建 db.py

db.py 文件是我们管理数据库连接和模型基类的地方。它让我们的代码更加模块化和可维护，实际生产中也是类似的，无论是在 FastAPI 或者 Flask 等框架中，当使用到 SqlAlchemy 时，的的确确需要一个单独 db.py，存储着引擎、会话以及模型基类。这个文件请务必建好，后续所有教程都会使用到。

from sqlalchemy import *
from sqlalchemy.orm import sessionmaker, declarative_base, Session, DeclarativeBase# 1.创建数据库驱动引擎
engine = create_engine(# url = '驱动://账户:密码@地址:端口/数据库名?charset=编码'url='mysql+pymysql://root:0908@localhost:3306/db_sqlalchemy_demo?charset=utf8mb4',echo=True,pool_size=8,  # 连接池的数据库连接数量max_overflow=30,  # 连接池的数据库连接最大数量pool_recycle=60 * 30,  # 设置秒数限制数据库多久没连接自动断开
)# 2.基于底层数据库驱动建立数据库连接会话
# DBSession = sessionmaker(bind=engine)
# session = DBSession()
session = Session(bind=engine)# 3.模型对象基类，提供数据库的基操和方法
# Model = declarative_base()
class Model(DeclarativeBase):pass

创建 model.py

该文件负责模型类创建，需要基于 db.py 文件中的各个对象来创建。方便于其他文件或者生产开发中的引入（比如 Flask 项目可以根据业务引入相对应的模型类，进行增删改查操作）

import datetime
import dbclass Student(db.Model):__tablename__ = 'tb_student'id = db.Column(db.Integer, primary_key=True, comment='学生编号')name = db.Column(db.String(20), comment='学生姓名')sex = db.Column(db.Boolean, default=True, comment='学生性别')age = db.Column(db.SmallInteger, comment='学生年龄')class_ = db.Column('class', db.SMALLINT, comment='学生班级')description = db.Column(db.Text, comment='个性签名')status = db.Column(db.Boolean, default=True, comment='登录状态')addtime = db.Column(db.DateTime, default=datetime.datetime.now, comment='入学时间')orders = db.Column(db.SMALLINT, default=True, comment='学生排序')def __repr__(self):return f'<{self.__class__.__name__}: {self.name}({self.id})>'def to_dict(self):return {'id': self.id,'name': self.name,'sex': self.sex,'age': self.age,'class': self.class_,'description': self.description,'status': self.status,'addtime': self.addtime.strftime('%Y-%m-%d %H:%M:%S'),'orders': self.orders,}

在此之前，确保数据库和表已经创建。如果没有创建，可以运行以下代码来创建表：

import dbdb.Model.metadata.create_all(db.engine)
# 删除所有表格如下操作：
db.Model.metadata.drop_all(db.engine)

添加数据

请注意，上方 db.py 以及 model.py 是在同一目录下的。

import db
from model import Studentdef run():try:student = Student(id=1,name='王小明',sex=True,  # 默认值为 True，可以省略age=18,class_=3,description='滚出去',status=True,  # 默认值为 True，可以省略orders=0  # 默认值为 0，可以省略)  # 实例化学生模型类db.session.add(student)  # 通过会话告知添加# 另外还有一个 add_all([student1, ...])db.session.commit()  # 我们使用的 MySQL 是支持事务操作的，上述实例化、添加无误后，可提交，这样才是真正的添加数据完毕print(student.to_dict())  # 打印瞅瞅except Exception as e:db.session.rollback()  # 出现异常就回滚（事务操作），并打印异常print(f"Error: {e}")finally:db.session.close()  # 最终手动关闭会话if __name__ == '__main__':run()

添加多条数据如下：

import db
from model import Studentstudents = [Student(name='汪伦', age=19, class_=3, description='滚出去', ),Student(name='上官丽丽', age=18, sex=False, class_=1, description='滚进来', ),
]
db.session.add_all(students)
db.session.commit()

查询数据

查询一条数据

现在我们来查询刚刚添加的数据：

import db
from model import Student# 调用会话的 Query 对象，该对象有一个 get 方法，传入主键值即可
# 查询 id 为 1 的学生，也就是 王小明
student = db.session.query(Student).get(1)  # 单个主键 联合主键则使用(1,2) 或者 {'id': 1, 'class_id': 2}
print(student.to_dict())
# 上述可能会提示警告，可使用下列方式
q1 = db.select(Student).where(Student.id == 1)  # SELECT * FROM Student WHERE id == 1
student = db.session.execute(q1).scalar()
print(student.to_dict())
# 该方式也行
q1 = db.select(Student)   # SELECT * FROM Student
student = db.session.execute(q1).scalar()  # 取出一条数据
print(student.to_dict())
# 这样也可以查询id为 1 的学生，方式还是很多的
student = db.session.query(Student).filter_by(id=1).first()
print(student.to_dict())
print(student.__dict__)

在后续的教程中，无非都是利用 db 中的 select、delete 等语句结合会话执行，又或者是会话中的 Query 对象进行筛选限制……

查询多条数据

import db
from model import Student# 获取所有的数据，其实和 SQL 语句还是有相似之处的
q2 = db.select(Student)    # SELECT * FROM Student
students = db.session.execute(q2).scalars()  # 创建迭代器对象
# scalar 是查询出来的首个，scalars 是所有，并不是所有数据，而是迭代器，遍历可拿到
# for stu in students:
#     print(stu.to_dict())# 务必注意以下这一点
print(students)  # 正常打印数据
print(students)  # 打印空列表 []
# 为什么？请参考迭代器的原理，简而言之就是（拿完了，没了~）
# 怎么解决？转成列表（赋值或者拷贝？未测试过，自行试试吧！）

条件过滤数据

import db
from model import Student# 官网示例的方式如下：
q3 = db.select(Student).where(Student.class_ == 3)
students = db.session.execute(q3).scalars()
for stu in students:print(stu.to_dict())
# 你也可以这样：
students = db.session.query(Student).filter_by(class_=3).all()
print(students)
students = db.session.query(Student).filter(Student.class_ == 3).all()
print(students)

值得注意的是 filter_by 不支持 大于小于等 范围查询，而需要使用 filter 或者官网示例中 Select 对象 where 语句进行查询。

逻辑查询数据

如果想使用并且、或者、取非这些逻辑条件怎么办呢？SQLAlchemy 可不支持我们 Python 语法的条件，你可以试试，是否报错或者结果有问题！如果数据没问题，我只能说巧了，少部分条件还是可以判断到的，但想要精确无误，还是需要使用下方的三个函数。

SQLAlchemy 内置有这三个相关的函数：and_、not_、or_，可相互嵌套使用。

1. and_(条件1, 条件2, ...)
2. not_(条件)
3. or_(条件1, 条件2, ...)

import db
from model import Student# 查询班级不是1的学生或者非女生，条件看似复杂，有内到外拆分
# 内部 and_ 表示 班级1的女生，外层是 not_，取反即可，班级1的女生除外，其他都能取
# 经过数学转义并更贴切表示就是：(not (班级 == 1)) or (not (性别 == 女))
q4 = db.select(Student).where(db.not_(db.and_(Student.class_ == 1, Student.sex == 0)))
students = db.session.execute(q4).scalars()
for stu in students:print(stu.to_dict())
# 三个条件函数可支持 select 对象中的 where 以及 会话 Query 对象中的 filter
students = db.session.query(Student).filter(db.not_(db.and_(Student.class_ == 1, Student.sex == 0))).all()
for stu in students:print(stu.to_dict())

到目前为止，我认为你有必要牢记两个对象，一个是 Select，源于 SqlAlchemy，另一个是 Qurey，源于我们的会话 session 中。Select 对象或者将来的 Delete 等，本质上可看做是构造 SQL 语句的，构造后通过会话的 execute 方法执行，有数据返回时可调用 scalar 或者 scalars 获取。而 Query 是查询过滤专用的，基于会话直接使用，可调用 all、first 等方法获取数据。

成员查询数据

除了上述的三个逻辑函数，还有一个模型对象的字段对象所拥有的方法：in_，也就是类似于我们 Python 中的成员运算符 in。SQLAlchemy 同样不支持 in 运算符，因此你还需要掌握字段对象的 in_ 方法

'a' in 'abcde'
# SQLAlchemy
student.id.in_([1, 3, 5, 6])

请参考实际例子：

import db
from model import Student# 查询学生编号为1、2的学生
students = db.session.query(Student).filter(Student.id.in_([1, 2])).all()
for stu in students:print(stu.to_dict())# 使用 Select 对象也可~
q5 = db.select(Student).where(Student.id.in_([1, 2]))
students = db.session.execute(q5).scalars()
for stu in students:print(stu.to_dict())

更新数据

查询到对象后，直接对对象属性进行赋值操作，可修改记录的字段值，再通过 commit 提交后，可保存记录。

import db
from model import Studentq = db.select(Student).where(Student.id == 1)
student = db.session.execute(q).scalar()
print(student.to_dict())  # 打印查询到的数据
student.name = '张晓明'  # 将 王小明 转成 张晓明
db.session.commit()  # 提交事务# 再此查询一遍
q = db.select(Student).where(Student.id == 1)
student = db.session.execute(q).scalar()
print(student.to_dict())  # 打印数据

删除数据

删除某条记录，首先需要有删除的条件，其次才能删除。

import db
from model import Student# 方式1
q = db.select(Student).where(Student.id == 1)
student = db.session.execute(q).scalar()
db.session.delete(student)
db.session.commit()
# 方式2
# q = db.delete(Student).where(Student.id == 1)
# db.session.execute(q)
# db.session.commit()# 查询一下数据
q = db.select(Student).where(Student.id == 1)
student = db.session.execute(q).scalar()
print(student)  # 应为 None

以上例子仅供参考，具体条件可以更加的灵活和简单。