Flask ORM 模型（轻松版）

ORM (Object-Relational Mapping) 是 Flask 中处理数据库的核心技术，它将数据库表映射为 Python 类，使开发者可以用面向对象的方式操作数据库。

Flask 最常用的 ORM 是 SQLAlchemy，通过 Flask-SQLAlchemy 扩展集成。

1. Flask-SQLAlchemy 基础配置

安装与初始化

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'  # SQLite 数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 禁用事件系统，减少内存使用
db = SQLAlchemy(app)

支持的数据库 URI 格式

• SQLite: sqlite:///database.db
• PostgreSQL: postgresql://user:password@localhost/mydatabase
• MySQL: mysql://user:password@localhost/mydatabase
• Oracle: oracle://user:password@localhost/mydatabase

2. 定义模型类

基本模型示例

class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)email = db.Column(db.String(120), unique=True, nullable=False)def __repr__(self):return f'<User {self.username}>'

字段类型

字段类型	描述
db.Integer	整数
db.String(size)	字符串，需指定最大长度
db.Text	长文本
db.DateTime	日期和时间
db.Float	浮点数
db.Boolean	布尔值
db.LargeBinary	二进制数据
db.Enum	枚举值

字段选项

选项	描述
primary_key	是否为主键
unique	是否唯一
nullable	是否允许为空
default	默认值
index	是否为该列创建索引
autoincrement	是否自动递增

3. 模型关系

一对多关系

class User(db.Model):id = db.Column(db.Integer, primary_key=True)posts = db.relationship('Post', backref='author', lazy=True)class Post(db.Model):id = db.Column(db.Integer, primary_key=True)user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

多对多关系

# 关联表
tags = db.Table('tags',db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True),db.Column('post_id', db.Integer, db.ForeignKey('post.id'), primary_key=True)
)class Post(db.Model):id = db.Column(db.Integer, primary_key=True)tags = db.relationship('Tag', secondary=tags, lazy='subquery',backref=db.backref('posts', lazy=True))class Tag(db.Model):id = db.Column(db.Integer, primary_key=True)

一对一关系

class User(db.Model):id = db.Column(db.Integer, primary_key=True)profile = db.relationship('Profile', backref='user', uselist=False)class Profile(db.Model):id = db.Column(db.Integer, primary_key=True)user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

4. 数据库操作

创建数据库表

with app.app_context():db.create_all()  # 创建所有定义的表

CRUD 操作

创建 (Create)

new_user = User(username='john', email='john@example.com')
db.session.add(new_user)
db.session.commit()

读取 (Read)

# 获取所有用户
users = User.query.all()# 获取单个用户
user = User.query.get(1)  # 通过主键获取# 条件查询
admin = User.query.filter_by(username='admin').first()# 复杂查询
recent_users = User.query.order_by(User.id.desc()).limit(5).all()

更新 (Update)

user = User.query.get(1)
user.email = 'new@example.com'
db.session.commit()

删除 (Delete)

user = User.query.get(1)
db.session.delete(user)
db.session.commit()

5. 高级查询技巧

链式调用

User.query.filter(User.username != 'admin')\.filter(User.email.like('%@gmail.com'))\.order_by(User.username)\.limit(10)\.all()

聚合函数

from sqlalchemy import func# 计数
count = db.session.query(func.count(User.id)).scalar()# 分组统计
result = db.session.query(User.username, func.count(Post.id))\.join(Post)\.group_by(User.username)\.all()

原生 SQL 查询

result = db.session.execute('SELECT * FROM user WHERE id = :id', {'id': 1})

6. 模型继承

单表继承

class Person(db.Model):id = db.Column(db.Integer, primary_key=True)type = db.Column(db.String(50))  # 鉴别器列__mapper_args__ = {'polymorphic_identity': 'person','polymorphic_on': type}class Employee(Person):__mapper_args__ = {'polymorphic_identity': 'employee'}salary = db.Column(db.Float)

多表继承

class Person(db.Model):id = db.Column(db.Integer, primary_key=True)type = db.Column(db.String(50))__mapper_args__ = {'polymorphic_identity': 'person','polymorphic_on': type}class Employee(Person):__tablename__ = 'employee'__mapper_args__ = {'polymorphic_identity': 'employee'}id = db.Column(db.Integer, db.ForeignKey('person.id'), primary_key=True)salary = db.Column(db.Float)

7. 钩子方法 (Hooks)

可以在模型中定义特殊方法来实现业务逻辑：

class User(db.Model):# ... 字段定义 ...def before_insert(self):self.created_at = datetime.utcnow()def after_update(self):self.updated_at = datetime.utcnow()# 注册事件监听
@event.listens_for(User, 'before_insert')
def before_insert_listener(mapper, connection, target):target.before_insert()@event.listens_for(User, 'after_update')
def after_update_listener(mapper, connection, target):target.after_update()

8. 性能优化

批量插入

users = [User(username=f'user{i}') for i in range(1000)]
db.session.bulk_save_objects(users)
db.session.commit()

延迟加载 vs 立即加载

# 延迟加载 (默认)
user = User.query.get(1)
posts = user.posts  # 此时才执行查询# 立即加载
user = User.query.options(db.joinedload(User.posts)).get(1)
posts = user.posts  # 已经预先加载

分页查询

page = request.args.get('page', 1, type=int)
per_page = 10
users = User.query.paginate(page=page, per_page=per_page)

9. 数据库迁移 (Flask-Migrate)

使用 Alembic 进行数据库迁移：

安装与配置

pip install flask-migrate

from flask_migrate import Migratemigrate = Migrate(app, db)

常用命令

flask db init          # 初始化迁移仓库
flask db migrate       # 创建迁移脚本
flask db upgrade       # 应用迁移
flask db downgrade     # 回滚迁移

10. 最佳实践

1. 分离模型定义：将模型放在单独的模块中（如 models.py）
2. 使用应用工厂模式：延迟数据库初始化
3. 合理使用事务：确保相关操作在一个事务中
4. 错误处理：正确处理数据库操作中的异常
5. 索引优化：为常用查询字段添加索引
6. 避免 N+1 查询问题：使用 joinedload 或 subqueryload
7. 定期维护：重建索引、清理碎片等

总结

Flask ORM 模型提供了强大而灵活的方式来处理数据库操作，通过 Flask-SQLAlchemy 可以：

• 用 Python 类表示数据库表
• 用对象属性表示表字段
• 用方法调用表示 SQL 操作
• 支持复杂的关系和查询
• 提供事务管理和连接池

掌握这些 ORM 技术可以让你在 Flask 应用中高效、安全地进行数据库操作，同时保持代码的整洁和可维护性。