Flask-SQLAlchemy数据库操作指南
ORM简介
ORM（Object-Relational Mapping）是对象关系映射的缩写，它能够将数据库表结构映射到程序中的模型类对象。通过实例化这些类并对其进行操作，可以间接完成对数据库表的操作。
使用ORM的主要优势包括：
- 简化数据库操作：无需手动编写复杂的SQL语句。
- 增强安全性：避免直接拼接SQL字符串，从而有效防止SQL注入攻击。
Python中常用的ORM框架有Django ORM和SQLAlchemy。其中，SQLAlchemy功能更为强大且灵活，支持多种数据库后端。
在Flask中集成SQLAlchemy
安装数据库驱动
无论使用哪种ORM框架，都需要先安装相应的数据库驱动程序。以MySQL为例，以下是常见的几种驱动：
mysqlclient：兼容性强，适合大多数场景。PyMySQL：纯Python实现，但性能稍逊。
例如，使用mysqlclient时可以通过以下命令安装：
pip install mysqlclient
创建模型类
下面是一个简单的例子，展示如何定义两个表（角色表和用户表）以及它们之间的一对多关系：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@127.0.0.1:3306/test_db'
db = SQLAlchemy(app)
class RoleModel(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True)
class UserModel(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True)
email = db.Column(db.String(100), unique=True)
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
role = db.relationship('RoleModel', backref='users')
数据库迁移
为了同步模型类与实际数据库结构，需要借助Flask-Migrate插件进行迁移操作。
安装Flask-Migrate
pip install flask-migrate
初始化迁移环境
首先设置环境变量：
export FLASK_APP=app.py
然后运行以下命令完成初始化：
flask db init
flask db migrate
flask db upgrade
测试数据操作
通过Flask Shell可以快速验证模型类的功能：
# 创建角色和用户
role_admin = RoleModel(name='admin')
db.session.add(role_admin)
db.session.commit()
user_xiaoming = UserModel(username='xiaoming', email='xiaoming@example.com', role_id=role_admin.id)
db.session.add(user_xiaoming)
db.session.commit()
# 查询关联关系
print(user_xiaoming.role.name) # 输出 'admin'
print(role_admin.users[0].username) # 输出 'xiaoming'
优化模型类
为提高可读性和易用性，可以重写__init__和__repr__方法：
class RoleModel(db.Model):
...
def __init__(self, name):
self.name = name
def __repr__(self):
return f'<Role {self.name}>'
class UserModel(db.Model):
...
def __init__(self, username, email, role):
self.username = username
self.email = email
self.role = role
def __repr__(self):
return f'<User {self.username}>'
CRUD操作示例
新增记录
new_role = RoleModel(name='editor')
db.session.add(new_role)
db.session.commit()
批量新增
users = [
UserModel(username='alice', email='alice@example.com', role=new_role),
UserModel(username='bob', email='bob@example.com', role=new_role)
]
db.session.add_all(users)
db.session.commit()
删除记录
target_role = RoleModel.query.filter_by(name='editor').first()
db.session.delete(target_role)
db.session.commit()
更新记录
方式一：修改属性值后提交：
target_user = UserModel.query.filter_by(username='alice').first()
target_user.email = 'new_alice@example.com'
db.session.add(target_user)
db.session.commit()
方式二：直接调用update方法：
UserModel.query.filter_by(username='bob').update({'email': 'new_bob@example.com'})
db.session.commit()
查询记录
常见查询方法：
all()：返回所有匹配项。first()：返回第一条匹配项。get(id)：根据主键ID获取单条记录。filter_by()：基于等值条件过滤。filter()：支持复杂条件过滤。
示例：
# 按条件查询
results = UserModel.query.filter(UserModel.email.endswith('@example.com')).all()
# 排序与分页
sorted_results = UserModel.query.order_by(UserModel.id.desc()).offset(1).limit(2).all()
# 分组统计
from sqlalchemy import func
grouped_results = db.session.query(UserModel.role_id, func.count(UserModel.role_id)).group_by(UserModel.role_id).all()
