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. 最佳实践
- 分离模型定义:将模型放在单独的模块中(如 models.py)
- 使用应用工厂模式:延迟数据库初始化
- 合理使用事务:确保相关操作在一个事务中
- 错误处理:正确处理数据库操作中的异常
- 索引优化:为常用查询字段添加索引
- 避免 N+1 查询问题:使用 joinedload 或 subqueryload
- 定期维护:重建索引、清理碎片等
总结
Flask ORM 模型提供了强大而灵活的方式来处理数据库操作,通过 Flask-SQLAlchemy 可以:
- 用 Python 类表示数据库表
- 用对象属性表示表字段
- 用方法调用表示 SQL 操作
- 支持复杂的关系和查询
- 提供事务管理和连接池
掌握这些 ORM 技术可以让你在 Flask 应用中高效、安全地进行数据库操作,同时保持代码的整洁和可维护性。