Flask 集成sqlalchemy

一、ORM框架介绍

1.框架概述
SQLAlchemy
- 类型：企业级ORM框架。
- 特点：功能强大，使用广泛，支持多种数据库后端。
- 适用性：可用于各种Python Web框架，包括Flask和FastAPI。
Python界的ORM框架
- Django ORM：专为Django框架设计，不适用于其他框架。
- Peewee：小型ORM框架，适合轻量级应用。
- SQLAlchemy：企业级，功能全面，适用于大型复杂应用。
- Tortoise ORM：较新，适用于FastAPI等现代异步框架。
同步框架与异步框架
- Django和Flask：传统同步框架，新版本开始支持异步特性。
- FastAPI、Sanic、Tornado：原生异步框架，支持异步编程模式，提高性能。
异步框架中的同步问题
- 在异步框架中使用同步第三方库可能导致性能瓶颈，因为异步框架的优势在于能够处理大量并发请求，而同步操作会阻塞事件循环。
- 为了充分利用异步框架的性能，推荐使用异步版本的第三方库，例如：
  - Redis：使用aioredis代替redis-py。
  - MySQL：使用aiomysql代替pymysql。
架构组成
- ```
[ Engine ]-----[ Connection Pooling ]|                              ^|                              ||-[ Dialect ]<-----------------+|
[ SQL Expression Language ]
```
- Engine 是 SQLAlchemy 的核心，负责创建会话和执行 SQL 语句。
- Connection Pooling 与 Engine 相连，显示它是 Engine 的一部分，用于管理数据库连接。
- Dialect 与 Engine 相连，表示 Engine 使用 Dialect 来配置和选择特定的数据库连接方式。
- SQL Expression Language 与 Engine 平行，显示它是 Engine 用来构建查询的一个独立但相关的组件。
2.简单使用
安装SQLAlchemy
- ```
pip install sqlalchemy==2.0.30
```

连接数据库

import pymysql
from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8",  # 创建数据库连接max_overflow=0,  # 设置最大溢出连接数pool_size=5,  # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)

连接不同数据库

PostgreSQL

# 默认 DB API:
engine = create_engine("postgresql://scott:tiger@localhost/mydatabase")
# 使用 psycopg2
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/mydatabase")
# 使用 pg8000
engine = create_engine("postgresql+pg8000://scott:tiger@localhost/mydatabase")

MySQL

# 默认 DB API
engine = create_engine("mysql://scott:tiger@localhost/foo")
# 使用 mysqlclient
engine = create_engine("mysql+mysqldb://scott:tiger@localhost/foo")
# 使用 PyMySQL
engine = create_engine("mysql+pymysql://scott:tiger@localhost/foo")

SQLite

# Unix/Mac
engine = create_engine("sqlite:absolute/path/to/foo.db")
# Windows
engine = create_engine("sqlite:///C:\\path\\to\\foo.db")

原生操作数据库
- 操作流程
  - 从 Engine 获取一个原生的数据库连接（raw_connection）。
  - 创建一个游标对象，准备执行 SQL 语句。
  - 执行 SELECT 语句，从 article 表中获取所有数据。
  - 使用 fetchall 获取查询结果并打印。
  - 关闭游标和数据库连接。
- ```
if __name__ == '__main__':conn = engine.raw_connection()cursor = conn.cursor(pymysql.cursors.DictCursor)cursor.execute("select * from user")result = cursor.fetchall()print(result)cursor.close()conn.close()
```

二.SQLalchemy操作

1.创建和删除表

创建基类：

一般创建一个models.py文件进行下列操作
创建基类，以后所有类，都必须继承基类

# 老版本创建基类(不建议)
# from sqlalchemy.ext.declarative import declarative_base
# Base = declarative_base()# 新版本创建基类
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):pass

定义模型：

写个类，继承基类

from sqlalchemy import Column, Integer, String, Text, DateTime, func
class User(Base):__tablename__ = 'user'  # 表名id = Column(Integer, primary_key=True, autoincrement=True)  # 主键自增name = Column(String(32), index=True, nullable=False)  # 索引不能为空email = Column(String(32), unique=True)  # 唯一索引ctime = Column(DateTime, default=func.now())  # 自动生成创建时间mtime = Column(DateTime, default=func.now(), onupdate=func.now())  # 自动更新修改时间extra = Column(Text, nullable=True)  # 额外字段def __repr__(self):  # 打印对象在别的容器内时显示的字符串return f"<User(id={self.id}, name={self.name}, email={self.email})>"def __str__(self):  # 打印对象时显示的字符串return f"User(id={self.id}, name={self.name}, email={self.email})"

创建/删除表：

使用基类Base.metadata.create_all(engine)来创建表。
使用基类Base.metadata.drop_all(engine)来删除表。

if __name__ == '__main__':from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1)# 创建表Base.metadata.create_all(engine)# 删除表Base.metadata.drop_all(engine)

2.字段增删改查

创建engine和session

首先要创建engine和session
还需要导入对应的模型

# 创建engine
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1
)# 创建 session对象
# # 老方式
# from sqlalchemy.orm import sessionmaker
# Session = sessionmaker(bind=engine)
# session=Session()
# 新方式(推荐)
from sqlalchemy.orm import Session
session = Session(engine)# 导入模型
from models import User

新增操作

首先使用模型类创建实例
通过session.add()或session.add_all()添加对象。
最后一定要提交事务

#  新增
user1 = User(name='bruce', email='1234@qq.com')
user2 = User(name='tom', email='5678@qq.com')
# session.add(user)    # 单个对象
session.add_all([user1, user2])  # 多个对象
session.commit()  # 提交
session.close()  # 关闭session

查询操作

使用session.query()来查询对象。

# 2 查看
res = session.get(User, 1)  # 通过id查询
res = session.query(User).count()  # 统计数量
res = session.query(User).first()  # 第一个对象
res = session.query(User).all()  # 所有对象
res = session.query(User).filter_by(name='bruce').all()  # 按条件查询
res = session.query(User).filter_by(id=1).all()  # 按id查询
res = session.query(User).filter(User.id > 1).all()  # 按条件查询
res = session.query(User.id, User.name).all()  # 指定字段查询
res = session.query(User.id, User.name).filter_by(id=1)  # 查看sql语句

更改操作

通过query().filter().update()来更新对象。
也可以查看单个对象，修改信息后在添加
最后一定要提交事务

session.query(User).filter_by(id=2).update({'name': 'liuliu'})
session.commit()
user = session.query(User).filter_by(id=2).first()
print(user.name)  # liuliu
user.name = 'haha'
session.add(user)  # 根据id判断是否存在，不存在则新增，存在则修改
session.commit()

删除操作

通过query().filter().delete()来删除对象。

session.query(User).filter_by(name='bruce').delete()  # 返回受影响的行数
session.commit()

三、scoped_session线程安全

scoped_session是SQLAlchemy提供的一种机制，用于保证在不同线程中使用Session时的线程安全性。

1.什么是scoped_session？
scoped_session是SQLAlchemy ORM会话工厂的一个装饰器，它会创建一个线程本地（thread-local）的Session实例。
这意味着每个线程都会拥有自己的Session实例，而不会与其他线程共享。
这样可以避免在并发请求中共享同一个Session实例时可能出现的线程安全问题。
2.简单使用scoped_session

创建Engine：创建一个Engine实例，它负责管理到数据库的连接。

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1
)

定义Session：使用sessionmaker创建一个Session类，然后通过scoped_session装饰器来确保线程安全。
```
from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker(bind=engine))
```

在应用中使用Session：通常会在请求开始时创建一个Session实例，并在请求结束时关闭它。

from flask import Flask, g
app = Flask(__name__)@app.before_request
def before_request():g.session = Session()@app.teardown_request
def teardown_request(exception):g.session.close()

在视图函数中使用Session：在视图函数中，可以直接从g对象中获取Session实例。

@app.route('/')
def index():from models import User# 进行数据库操作user = g.session.query(User).filter_by(name='bruce').first()return f'User info：{user}'

3.线程安全的意义

如果使用全局Session实例，那么当多个请求同时进行数据库操作时，可能会出现以下问题：

线程安全问题：多个线程同时访问和修改同一个Session实例，可能导致数据不一致。
事务边界不清：由于多个请求共享同一个Session，事务的开启和提交可能会变得混乱，导致难以追踪和调试。

四、表关系一对多

1 关系定义

在SQLAlchemy中，一对多关系通过relationship函数和ForeignKey约束来定义。

Hobby类：代表一个爱好，它是一个独立的实体，拥有多个喜欢这个爱好的人。
Person类：代表一个人，他/她可以有一个爱好。

以下是这两个类的定义：

# 新版本创建基类
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):passfrom sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationshipclass Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)desc = Column(String(50), default='篮球')def __repr__(self):return f"<Hobby(id={self.id}, caption={self.desc})>"def __str__(self):return self.descclass Person(Base):__tablename__ = 'person'id = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)hobby_id = Column(Integer, ForeignKey("hobby.id"))  # 定义外键，关联到hobby表的id字段，实际字段hobby = relationship('Hobby', backref='pers')  # 定义关系，关联到Hobby类，并设置反向引用为pers，非实际字段def __repr__(self):return f"<Person(id={self.id}, name={self.name}, hobby={self.hobby.desc})>"def __str__(self):return self.nameif __name__ == '__main__':from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1)# 创建表Base.metadata.create_all(engine)

2.关系操作

创建Engine和Session

在开始操作数据库之前，需要创建一个Engine实例和Session对象

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import Person, Hobby
engine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1
)
session = Session(engine)

新增数据

简便方法：在添加Person对象时直接关联一个Hobby对象。
分步方法：先添加Hobby，然后添加Person并指定hobby_id。
直接添加：直接添加Person对象并关联Hobby对象

# 在添加`Person`对象时直接关联一个`Hobby`对象
# person = Person(name='bruce', hobby=Hobby(desc='羽毛球'))
# session.add(person)
# session.commit()# 先添加`Hobby`，然后添加`Person`并指定`hobby_id`
# hobby = Hobby(desc='足球')
# session.add(hobby)
# session.commit()
# person = Person(name='alice', hobby_id=hobby.id)
# session.add(person)
# session.commit()# 直接添加`Person`对象并关联`Hobby`对象
hobby = Hobby(desc='排球')
person = Person(name='bob', hobby=hobby)
session.add(person)
session.commit()

查询数据

正向查询：通过Person对象查询其关联的Hobby。
反向查询：通过Hobby对象查询所有喜欢这个爱好的人。

# 正向查询
# person = session.query(Person).filter_by(id=2).first()
# print(person.hobby.desc)
# 反向查询
hobby = session.query(Hobby).filter_by(id=1).first()
for p in hobby.pers:  # 使用backref定义的属性进行查询print(p.name)

修改数据

修改Person和Hobby之间的关系

# 更改关系
person = session.query(Person).filter_by(id=1).first()
new_hobby = session.query(Hobby).filter_by(id=2).first()
if person and new_hobby:person.hobby = new_hobby  # 修改关系session.commit()  # 提交修改

删除数据

删除Person对象

# 查询并删除Person对象
person = session.query(Person).filter_by(id=1).first()
if person:# 删除Person对象session.delete(person)# 提交删除操作session.commit()

删除Hobby对象

# 查询并删除Hobby对象
hobby = session.query(Hobby).filter_by(id=1).first()
if hobby:# 删除Hobby之前也删除所有关联的Person对象for person in hobby.pers:session.delete(person)# 删除Hobby对象session.delete(hobby)# 提交删除操作session.commit()
如果不想删除关联的Person对象，需要确保在Person表中外键字段是可选的（nullable=True）
希望在删除Hobby对象时自动删除所有关联的Person对象，可以在定义关系时使用cascade参数。
class Person(Base):# ...hobby = relationship('Hobby', backref='pers', cascade='all, delete, delete-orphan')# ...

五、表关系多对多

1.关系定义

多对多关系的创建是通过定义三个模型类来实现的：两个主体模型和一个关联表

创建基类表

from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):pass

定义关联表

关联表包含了两个外键，分别指向两个主体表的主键。

from sqlalchemy import Column, Integer, String, ForeignKey, func, DateTime
from sqlalchemy.orm import relationshipclass Boy2Girl(Base):__tablename__ = 'boy2girl'id = Column(Integer, primary_key=True, autoincrement=True)girl_id = Column(Integer, ForeignKey('girl.id'))boy_id = Column(Integer, ForeignKey('boy.id'))ctime = Column(DateTime, default=func.now())

id 是关联表的主键，通常不需要与任何实体表的主键关联。
girl_id 和 boy_id 是外键，分别引用 Girl 和 Boy 表的主键。

定义主体模型

两个主体模型，它们通过 relationship 函数与关联表建立多对多关系

class Girl(Base):__tablename__ = 'girl'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)def __repr__(self):return f'<Girl(id={self.id}, name={self.name})>'def __str__(self):return self.nameclass Boy(Base):__tablename__ = 'boy'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(64), unique=True, nullable=False)girls = relationship('Girl', secondary='boy2girl', backref='boys') #def __repr__(self):return f'<Boy(id={self.id}, name={self.name})>'def __str__(self):return self.name

在 Boy 类中，girls 属性是一个多对多的关系。它通过 secondary 参数指向关联表的名称（boy2girl），这样 SQLAlchemy 就知道如何连接这些表。
backref 参数在 Girl 类中创建了一个反向引用，允许从 Girl 对象访问与之关联的所有 Boy 对象。

创建表

if __name__ == '__main__':from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1)# 创建表Base.metadata.create_all(engine)

2.关系操作

首先创建engine和session，并导入模型表

Engine 负责管理数据库连接。
Session 是一个事务性接口，用于执行数据库操作。

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import Girl, Boyengine = create_engine("mysql+pymysql://root:000@127.0.0.1:3306/test?charset=utf8mb4",max_overflow=0,pool_size=5,pool_timeout=30,pool_recycle=-1
)
session = Session(engine)

增加记录

可以在创建是直接添加
可以通过第三章表关系添加
可以使用extend添加多个

# 直接添加
# girl = Girl(name="小红")
# boy = Boy(name="小刚", girls=[girl])
# session.add_all([girl, boy])
# session.commit()# 通过第三章表关系添加
# girl = Girl(name="李华")
# boy = Boy(name="小蓝")
# boy.girls.append(girl)
# session.add_all([girl, boy])
# session.commit()# extend方法添加
girls = session.query(Girl).all()
boy = Boy(name="小白")
boy.girls.extend(girls)
session.add_all([boy])
session.commit()

查询

主要分为正向查询和反向查询

# 正向查询
boy = session.query(Boy).filter_by(id=2).first()
res = boy.girls
print(res)# 反向查询
girl = session.query(Girl).filter_by(id=2).first()
res = girl.boys
print(res)

修改

修改多对多关系通常涉及到添加或删除关联。

# 修改关系
boy = session.query(Boy).filter_by(id=2).first()
girl = session.query(Girl).filter_by(id=2).first()
boy.girls.remove(girl)   # 移除关系
session.commit()
print(boy.girls)
boy.girls.append(girl)   # 添加关系
session.commit()
print(boy.girls)

删除

在定义模型时，可以通过 relationship 的 cascade 参数来设置级联行为。

# 删除
boy = session.query(Boy).filter_by(id=2).first()
session.delete(boy)
session.commit()

六、常用查询

filter_by 查询

filter_by 是一种简单的查询方法，可以直接通过指定属性名和值进行筛选。

# 查询 name 为 'bruce' 且 id 为 4 的用户
res = session.query(User).filter_by(name='bruce', id=4).all()
# 查询 name 为 'lucy' 的所有用户
res = session.query(User).filter_by(name='lucy').all()
# 查询 name 为 'kan' 的第一个用户
res = session.query(User).filter_by(name='kan').first()

filter[where] 查询

filter 或 where 用于更复杂的条件查询。

# 查询 id 大于等于 1 的所有用户
res = session.query(User).where(User.id >= 1).all()  # where 是 filter 的别名
res = session.query(User).filter(User.id >= 1).all()

between 查询

用于查询在指定范围内的记录。

# 查询 id 在 1 到 38 之间的用户
res = session.query(User).where(User.id.between(1, 10)).all()

in 查询

用于查询在指定列表中的记录。

# 查询 id 在 [1, 3, 5] 列表中的用户
res = session.query(User).filter(User.id.in_([1, 3, 5])).all()

非（~）查询

用于查询不在指定列表中的记录。

# 查询 id 不在 [1, 3, 4] 列表中的用户
res = session.query(User).filter(~User.id.in_([1, 3, 4])).all()

二次筛选

用于基于子查询的结果进行筛选

# 查询 id 在 (select id from user where name=xxx) 子查询结果中的用户
res = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='xxx'))).all()

and, or 条件

用于组合多个查询条件。

from sqlalchemy import and_, or_
# 查询 id > 3 且 name 为 'bruce' 的用户
res = session.query(User).filter(and_(User.id > 3, User.name == 'bruce')).all()
# 查询 id < 2 或 name 为 'bruce' 的用户
res = session.query(User).filter(or_(User.id < 2, User.name == 'bruce')).all()

通配符查询

用于模糊查询

# 查询 name 以 'b%' 开头的用户
res = session.query(User).filter(User.name.like('b%')).all()# 查询 name 不以 'b%' 开头的用户
ret = session.query(User).filter(~User.name.like('b%'))

限制（Limit）查询

用于分页

# 获取第一页，一页显示三条记录
ret = session.query(User)[0:3]

排序（Order By）

用于对结果进行排序。

# 按照名字升序排列
ret = session.query(User).order_by(User.name.asc()).all()# 按照名字降序排列，如果名字相同，则按照 id 升序排列
ret = session.query(User).order_by(User.name.desc(), User.id.asc()).all()

分组（Group By）

用于对结果进行分组

from sqlalchemy.sql import func
# 按照名字分组
ret = session.query(User).group_by(User.name).all()
# 分组后取最大 id，id 之和，最小 id
ret = session.query(func.max(User.id), func.sum(User.id), func.min(User.id), User.name).group_by(User.name).all()

连表查询

用于查询多个表。

# 笛卡尔积查询，然后根据条件过滤
ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all()# 内连接查询
ret = session.query(Person).join(Hobby)# 左外连接查询
ret = session.query(Person).join(Hobby, isouter=True)

组合查询（Union）

用于合并多个

# 使用 union 合并两个查询的结果集，会去除重复的行
q1 = session.query(Person.name)
q2 = session.query(Hobby.desc)
ret = q1.union(q2).all()# 使用 union_all 合并两个查询的结果集，不会去除重复的行
ret = q1.union_all(q2).all()

七、集成使用

1.安装必要包

pip install Flask-SQLAlchemy
# 管理数据库的迁移
pip install Flask-Migrate

2.导入和初始化
在app.py文件导入和初始化

from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config.from_pyfile('settings.py')  # 加载配置文件
db = SQLAlchemy(app)
migrate = Migrate(app, db)  # 初始化 Flask-Migrate

3.添加配置
在settings.py文件添加配置内容

SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:password@127.0.0.1:3306/your_database?charset=utf8"SQLALCHEMY_POOL_SIZE = 5  # 连接池大小
SQLALCHEMY_POOL_TIMEOUT = 30  # # 连接池超时时间
SQLALCHEMY_POOL_RECYCLE = -1  # 连接池回收时间，-1表示永不回收
SQLALCHEMY_TRACK_MODIFICATIONS = False  # 是否追踪对象的修改

4.定义模型
在你的 models.py 文件中，定义模型类并继承自 db.Model。

from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()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 '<User %r>' % self.usernamedef __str__(self):return self.username

5.使用会话
在视图函数中，你可以使用 db.session 来执行数据库操作。

from flask import Flask, request, jsonify
from models import User@app.route('/users', methods=['POST'])
def create_user():data = request.get_json()new_user = User(username=data['username'], email=data['email'])db.session.add(new_user)db.session.commit()return jsonify({'message': 'User created successfully'}), 201

6.数据迁移
初始化迁移环境
- 只需要执行一次，它会在你的项目中创建迁移环境。
- ```
flask db init
```
- 这个命令会在项目中创建一个 migrations 目录，这个目录包含了迁移脚本所需的所有配置文件。
自动生成迁移脚本
- ```
flask db migrate -m "Initial migration."
```
- 这个命令会自动检测你的模型变化，并生成一个新的迁移脚本。参数 -m 是对这次迁移的描述。
应用迁移
- ```
flask db upgrade
```
- 这个命令会应用迁移脚本到数据库中，更新数据库结构。
回滚迁移
- 如果你需要回滚到之前的版本
- ```
flask db downgrade
```
注意事项：
- 在使用 Flask-Migrate 命令之前，请确保已经定义了模型并在配置文件中设置了正确的数据库 URI。
- flask db init 只需要执行一次，它会在项目中创建迁移环境。
- 在生成迁移脚本时，确保已经对模型进行了所有需要的修改，因为迁移脚本是基于模型当前状态和数据库当前状态之间的差异生成的。
- 在应用迁移之前，最好备份数据库，以防万一迁移过程中出现问题。
- Flask-Migrate 支持命令行参数 --app 和 :app，这允许在多个 Flask 应用程序的项目中指定要使用哪个应用程序。

Flask 集成sqlalchemy