当前位置：首页 > news >正文

Flask中ORM的使用

news 2025/8/15 6:08:32

Flask中ORM的使用

本文介绍Flask中ORM框架flask_sqlalchemy的基本使用，包含模型定义(简单模型，一对一，一对多，多对多等)，由于实际开发中很少使用物理外键，所有本文所有模型都不使用物理外键，而关联关系db.relationship是应用层面的，故仍然使用。以下模型以学生、课程、班级为例，这三者基本涵盖了所有的模型对应关系。

模型定义（不使用物理外键）

1. 基础模型

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.associationproxy import association_proxydb = SQLAlchemy()class Student(db.Model):"""学生模型"""__tablename__ = 'students'id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(80), nullable=False)class_id = db.Column(db.Integer)  # 不使用物理外键# 关系属性class_rel = db.relationship('Class',primaryjoin='Class.id == Student.class_id',viewonly=True,back_populates='students_rel')# 关联代理class_name = association_proxy('class_rel', 'name')def __repr__(self):return f'<Student {self.name}>'class Class(db.Model):"""班级模型"""__tablename__ = 'classes'id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(50), unique=True, nullable=False)# 关系属性students_rel = db.relationship('Student',primaryjoin='Class.id == Student.class_id',viewonly=True,back_populates='class_rel')# 关联代理students = association_proxy('students_rel', 'self')student_names = association_proxy('students_rel', 'name')def __repr__(self):return f'<Class {self.name}>'class Course(db.Model):"""课程模型"""__tablename__ = 'courses'id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100), unique=True, nullable=False)credit = db.Column(db.Integer, default=1)def __repr__(self):return f'<Course {self.name}>'

2. 多对多关联模型

# 学生-课程关联表（不使用物理外键）
class StudentCourse(db.Model):"""学生-课程关联模型"""__tablename__ = 'student_courses'id = db.Column(db.Integer, primary_key=True)student_id = db.Column(db.Integer, nullable=False)course_id = db.Column(db.Integer, nullable=False)score = db.Column(db.Float)  # 成绩# 学生关系student_rel = db.relationship('Student',primaryjoin='Student.id == StudentCourse.student_id',viewonly=True)# 课程关系course_rel = db.relationship('Course',primaryjoin='Course.id == StudentCourse.course_id',viewonly=True)# 关联代理student_name = association_proxy('student_rel', 'name')course_name = association_proxy('course_rel', 'name')def __repr__(self):return f'<StudentCourse student:{self.student_id} course:{self.course_id}>'# 为Student添加课程关系
Student.courses = db.relationship('StudentCourse',primaryjoin='Student.id == StudentCourse.student_id',backref='student',viewonly=True
)# 为Course添加学生关系
Course.students = db.relationship('StudentCourse',primaryjoin='Course.id == StudentCourse.course_id',backref='course',viewonly=True
)

查询操作

1. 简单查询（带过滤条件）

# 查询所有学生
all_students = Student.query.all()# 查询学分大于2的课程
high_credit_courses = Course.query.filter(Course.credit > 2).all()# 查询姓"张"的学生
zhang_students = Student.query.filter(Student.name.like('张%')).all()# 分页查询班级
page = Class.query.paginate(page=1, per_page=10, error_out=False)

2. 一对一关系查询（学生-班级）

使用 relationship 方式：

# 查询学生及其班级名称
student = Student.query.get(1)
print(f"学生: {student.name}, 班级: {student.class_name}")# 查询班级及其所有学生
class_obj = Class.query.get(101)
for student in class_obj.students:print(f"班级 {class_obj.name} 的学生: {student.name}")

使用 JOIN 方式：

# 查询学生及其班级信息
result = db.session.query(Student, Class.name)\.join(Class, Student.class_id == Class.id)\.filter(Student.id == 1)\.first()if result:student, class_name = resultprint(f"学生: {student.name}, 班级: {class_name}")# 查询班级及其学生数量
from sqlalchemy import funcclass_info = db.session.query(Class.name,func.count(Student.id).label('student_count')
).join(Student, Class.id == Student.class_id).group_by(Class.id).all()

3. 一对多关系查询（班级-学生）

使用 relationship 方式：

# 查询班级及其所有学生
class_obj = Class.query.get(101)
print(f"班级: {class_obj.name}")
for student in class_obj.students:print(f" - 学生: {student.name}")# 使用预加载优化
classes = Class.query.options(db.joinedload(Class.students_rel)).all()

使用 JOIN 方式：

# 查询班级及其学生
results = db.session.query(Class.name, Student.name)\.join(Student, Class.id == Student.class_id)\.filter(Class.id == 101)\.all()for class_name, student_name in results:print(f"班级: {class_name}, 学生: {student_name}")# 查询每个班级的学生数量
class_counts = db.session.query(Class.name,func.count(Student.id).label('count').outerjoin(Student, Class.id == Student.class_id).group_by(Class.id).order_by(db.desc('count')).all()

4. 多对多关系查询（学生-课程）

使用 relationship 方式：

# 查询学生的所有课程
student = Student.query.get(1)
for sc in student.courses:  # sc 是 StudentCourse 对象print(f"课程: {sc.course_name}, 成绩: {sc.score or '未录入'}")# 查询课程的所有学生
course = Course.query.get(201)
for sc in course.students:print(f"学生: {sc.student_name}, 成绩: {sc.score or '未录入'}")# 使用关联代理直接获取课程名称
student = Student.query.get(1)
course_names = [sc.course_name for sc in student.courses]
print(f"学生 {student.name} 的课程: {', '.join(course_names)}")

使用 JOIN 方式：

# 查询学生及其课程成绩
results = db.session.query(Student.name, Course.name, StudentCourse.score)\.join(StudentCourse, Student.id == StudentCourse.student_id)\.join(Course, Course.id == StudentCourse.course_id)\.filter(Student.id == 1)\.all()for student_name, course_name, score in results:print(f"学生: {student_name}, 课程: {course_name}, 成绩: {score}")# 查询每门课程的选修人数
course_stats = db.session.query(Course.name,func.count(StudentCourse.student_id).label('student_count'),func.avg(StudentCourse.score).label('avg_score'))\.join(StudentCourse, Course.id == StudentCourse.course_id)\.group_by(Course.id)\.order_by(db.desc('student_count'))\.all()

新增操作

1. 简单模型的单个新增

# 新增班级
new_class = Class(name="计算机科学2023级")
db.session.add(new_class)
db.session.commit()# 新增学生
new_student = Student(name="张三", class_id=new_class.id)
db.session.add(new_student)
db.session.commit()# 新增课程
new_course = Course(name="数据库原理", credit=3)
db.session.add(new_course)
db.session.commit()

2. 简单模型的批量新增

# 批量新增学生
students_data = [{"name": "李四", "class_id": new_class.id},{"name": "王五", "class_id": new_class.id},{"name": "赵六", "class_id": new_class.id}
]students = [Student(**data) for data in students_data]
db.session.add_all(students)
db.session.commit()# 批量新增课程
courses_data = [{"name": "数据结构", "credit": 4},{"name": "算法设计", "credit": 3},{"name": "操作系统", "credit": 4}
]courses = [Course(**data) for data in courses_data]
db.session.add_all(courses)
db.session.commit()

3. 关联关系新增

# 学生选课（添加多对多关系）
# 获取学生和课程
student = Student.query.filter_by(name="张三").first()
course1 = Course.query.filter_by(name="数据库原理").first()
course2 = Course.query.filter_by(name="数据结构").first()# 添加选课记录
sc1 = StudentCourse(student_id=student.id, course_id=course1.id, score=92.5)
sc2 = StudentCourse(student_id=student.id, course_id=course2.id)db.session.add_all([sc1, sc2])
db.session.commit()

修改操作

1. 简单模型的修改

# 修改学生信息
student = Student.query.get(1)
if student:student.name = "张三丰"  # 修改姓名db.session.commit()# 修改课程学分
Course.query.filter_by(name="数据库原理").update({"credit": 4})
db.session.commit()

2. 条件修改

# 为所有2023级班级的学生增加学分（假设有class_year字段）
# 先找到2023级班级
class_2023 = Class.query.filter(Class.name.like("%2023级")).all()
class_ids = [c.id for c in class_2023]# 为这些班级的所有学生增加一门选修课
new_course = Course.query.filter_by(name="人工智能导论").first()if new_course:# 找出这些班级中还没选修该课程的学生students = Student.query.filter(Student.class_id.in_(class_ids),~Student.id.in_(db.session.query(StudentCourse.student_id).filter(StudentCourse.course_id == new_course.id))).all()# 为这些学生添加选课记录new_records = [StudentCourse(student_id=s.id, course_id=new_course.id)for s in students]db.session.add_all(new_records)db.session.commit()

删除操作

1. 简单模型的单个删除

# 删除一个学生
student = Student.query.get(1)
if student:# 先删除关联的选课记录StudentCourse.query.filter_by(student_id=student.id).delete()# 再删除学生db.session.delete(student)db.session.commit()# 删除一门课程
course = Course.query.get(201)
if course:# 先删除关联的选课记录StudentCourse.query.filter_by(course_id=course.id).delete()# 再删除课程db.session.delete(course)db.session.commit()

2. 简单模型的批量删除

# 删除所有没有选修任何课程的学生
# 先找出没有选课的学生
no_course_students = Student.query.filter(~Student.id.in_(db.session.query(StudentCourse.student_id))
).all()# 批量删除
for student in no_course_students:db.session.delete(student)db.session.commit()# 删除空班级
empty_classes = db.session.query(Class)\.outerjoin(Student, Class.id == Student.class_id)\.group_by(Class.id)\.having(func.count(Student.id) == 0)\.all()for class_obj in empty_classes:db.session.delete(class_obj)db.session.commit()

最佳实践总结

1. 模型设计建议

明确关系类型：准确区分一对一、一对多、多对多关系
使用关联代理：简化关联属性访问，如student.class_name
添加索引：对经常查询的字段添加索引
避免循环导入：将模型定义放在单独文件中

2. 查询优化技巧

N+1问题：始终使用joinedload或selectinload预加载关联数据
按需加载：使用load_only限制返回字段
批量操作：优先使用批量查询和操作，减少数据库交互次数
分页处理：对大数据集使用分页查询

3. 事务管理

try:# 执行数据库操作db.session.commit()
except Exception as e:db.session.rollback()# 处理异常
finally:db.session.close()

4. 性能监控

# 启用SQL日志
app.config['SQLALCHEMY_ECHO'] = True# 使用EXPLAIN分析慢查询
slow_query = Student.query.filter(Student.name.like('张%'))
explain = db.session.execute(f"EXPLAIN ANALYZE {str(slow_query.statement)}")
for line in explain:print(line[0])