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])
5. 安全注意事項
- 參數化查詢:始終使用ORM查詢或參數化SQL,防止SQL注入
- 數據驗證:在提交前驗證所有輸入數據
- 權限控制:實現細粒度的數據訪問控制
通過本指南,您應該能夠全面掌握在不使用物理外鍵的情況下,如何設計ORM模型并執行各種數據庫操作。學生-班級-課程的示例覆蓋了大多數實際開發場景,可作為您項目開發的參考模板。