Flask连接MySQL与ORM增删改查：从配置到实战踩坑全解析

Flask连接MySQL数据库，加上ORM增删改查，这套组合几乎是每个Flask后端新手都要迈过的坎。我这两年带新人、写实战项目，发现大家卡住的地方高度一致：数据库辛辛苦苦连上了，结果增删改查的代码要么写得又臭又长，要么动不动就出现中文乱码、事务不提交、数据悄悄丢了的怪问题。这篇内容就是把"Flask连接MySQL数据库+ORM增删改查"这件事从头到尾捋一遍，不只是贴代码，还会解释每个环节为什么这么设计、哪些地方最容易踩坑。

先说清楚这篇内容的适用范围。它适合刚开始接触Flask后端、准备把数据库接进项目的同学，也适合已经从原生SQL切到ORM但用得不顺手、想系统梳理一遍的开发者。我会从环境准备、连接配置、模型定义、增删改查四件套、实测踩坑这几个维度展开，最后补一点项目实践里沉淀下来的经验。整篇内容可以直接照着操作，也能作为平时写代码的参考手册。

1. 动手前的准备：环境选择与依赖安装

1.1 MySQL版本与Flask生态的匹配问题

很多人安装完MySQL就不管了，直接pip install flask然后开始写，结果在连接阶段各种报错。先说MySQL这边。目前主流是MySQL 5.7和8.0两个大版本，8.0默认的认证插件是caching_sha2_password，而5.7用的还是mysql_native_password。这个差异直接影响Python驱动能不能连上。

如果你用pymysql连接MySQL 8.0，老版本的pymysql会出现Authentication plugin 'caching_sha2_password' cannot be loaded这类的报错。解决办法有两个：要么把pymysql升级到比较新的版本，要么在MySQL里把用户的认证插件改回mysql_native_password。我在实际项目中更倾向于升级pymysql，因为改认证插件始终是动数据库配置，线上环境里这一下可能引发其他兼容问题。

Flask版本这里也顺带说一句。Flask 2.x和3.x目前都在活跃使用，对应的Flask-SQLAlchemy建议用3.x版本，API上有些变化，比如db.session.get()取代了老旧的query.get()，后面讲查询的时候会专门提到。建议环境组合是：

1.2 依赖装齐：不只是Flask-SQLAlchemy

安装命令很简单，但背后有几个点值得说清楚：

pip install flask flask-sqlalchemy pymysql cryptography

为什么装pymysql？因为Flask-SQLAlchemy本身不直接连数据库，它依赖SQLAlchemy，而SQLAlchemy需要对应的数据库驱动才能跟MySQL通信。常见的MySQL Python驱动有三个：PyMySQL、mysqlclient、mysql-connector-python。PyMySQL是纯Python实现的，安装最省心，跨平台都没问题；mysqlclient是C扩展，性能好一些，但在Windows上编译很容易翻车。综合考虑，本地开发和中小型项目直接用PyMySQL完全够用。

安装完成后，可以快速验证一下：

import pymysql import flask import flask_sqlalchemy print(pymysql.__version__, flask.__version__, flask_sqlalchemy.__version__)

这里有个容易忽略的点：cryptography这个包。MySQL 8.0的caching_sha2_password认证在有些场景下需要cryptography库来做加密通信，如果不装，连接时可能报RuntimeError: 'cryptography' package is required for sha256_password or caching_sha2_password auth methods。搜索热词里就出现过"mysql ssl连接错误"，后面第5章我会专门讲这个报错的处理。

2. 从原生SQL到ORM：连接配置背后的设计逻辑

2.1 数据库连接串的构成拆解

在Flask里配置MySQL连接，核心就是配置一个连接串，术语叫Database URL。它长这样：

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@127.0.0.1:3306/flask_demo?charset=utf8mb4'

很多人直接把这段配置抄走，改个用户名密码就完事，但连接串每一段都是什么意思、能不能动，应该搞清楚。把这段拆开看：

• mysql+pymysql：这是方言（dialect）+驱动（driver）的组合，意思是"我要用MySQL数据库，驱动力用PyMySQL"。如果这里写mysql+mysqlconnector，就是换用mysql-connector-python驱动了。
• root:123456：冒号左边是用户名，右边是密码。
• @127.0.0.1:3306：数据库所在的主机地址和端口，本地开发就是127.0.0.1，默认端口3306。
• /flask_demo：要连接的数据库名，前提是这个数据库已经创建好了。
• ?charset=utf8mb4：连接参数，这里指定了通信字符集是utf8mb4。

连接串里最容易翻车的是密码。如果密码里带了@、:、/这些特殊字符，直接拼进去URL解析就乱了。我自己遇到过密码里含@的情况，调了半天，最后发现需要做URL编码。Python里有现成的工具：

from urllib.parse import quote_plus password = quote_plus('p@ss:word') uri = f'mysql+pymysql://root:{password}@127.0.0.1:3306/flask_demo?charset=utf8mb4'

2.2 SQLAlchemy的三层核心机制：Engine、Session与Model

配置好连接串，初始化一个SQLAlchemy对象：

from flask import Flask from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@127.0.0.1:3306/flask_demo?charset=utf8mb4' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db = SQLAlchemy(app)

很多人以为这行SQLAlchemy(app)执行完就连接数据库了，其实没有。SQLAlchemy是延迟加载的，真正建立连接发生在第一次执行SQL语句的时候。这个机制背后是三个核心组件在协作：

• Engine（引擎）：由连接串创建，负责管理数据库连接池。它不执行业务逻辑，只是一个"连接调度中心"。
• Session（会话）：可以理解成"工作单元"。你在Session里做增删改查，Session负责把这些操作翻译成SQL，并维护事务状态。
• Model（模型）：用Python类描述表结构，ORM的核心，后面专门讲。

用生活化的比喻理解：Engine像是打车平台的总调度室，维护着一批空闲车辆（连接）；Session是你叫到的那辆具体出租车，你告诉司机去哪儿（执行查询），到目的地该付钱（commit）就付钱；Model则是地图上的地点标注，告诉你某个坐标对应一个什么位置。

Flask-SQLAlchemy比原生SQLAlchemy多做的事是：它在应用层帮你管理了一个"请求级"的Session作用域。也就是每个请求进来会有一个独立的Session，请求结束自动清理。这也是为什么在视图函数里可以直接用db.session而不需要手动创建。

3. 定义数据模型：一张表在ORM里长什么样

3.1 Model类的字段映射规则

数据库里的表，在ORM里就是一个继承db.Model的Python类。类属性对应字段，一个实例对应一行记录。以最典型的用户表为例：

from datetime import datetime class User(db.Model): __tablename__ = 'user' # 显式指定表名 id = db.Column(db.Integer, primary_key=True, autoincrement=True) username = db.Column(db.String(50), nullable=False, unique=True, index=True) email = db.Column(db.String(120), nullable=False) age = db.Column(db.Integer, default=0) is_active = db.Column(db.Boolean, default=True) created_at = db.Column(db.DateTime, default=datetime.now) def __repr__(self): return f'<User {self.username}>'

__tablename__不写的话，SQLAlchemy默认会按类名自动生成表名，但自动生成的表名在多词类名上经常不符合你的预期。比如UserProfile会自动转成user_profile，虽然也不差，但显式声明是更好的习惯。

字段类型这块，如果之前写过原生SQL建表，其实是一一对应的关系：

字段的约束参数也要养成习惯：nullable=False表示非空，unique=True表示唯一索引，default是Python侧的默认值，index=True会自动帮你建索引。等项目里表字段多了，这些约束就是数据质量的第一道防线。

3.2 db.create_all()还是用迁移工具建表？

模型定义好之后，需要把表建到MySQL里。最省事的办法：

with app.app_context(): db.create_all()

这段代码会在数据库里自动创建所有尚未存在的表。但它有一个大问题：只能在开发环境用来快速起表。如果线上数据库已经有表了，或者模型字段加了新列，create_all()不会帮你更新已有的表结构，它只会建不存在的表。换句话说，它不是迁移工具，不具备alter table的能力。

我见过有人把create_all()放在项目启动代码里，每次重启都执行一次，开发阶段没什么问题，但一旦上了生产环境，字段变更就完全失控了。正规做法是引入Flask-Migrate或Alembic，通过迁移脚本管理表结构变更。有同学觉得迁移工具麻烦，但对稍微正式点的项目，迁移脚本记录表结构变化历史的能力，在排查问题时实在太宝贵了。

如果是手写SQL建表，那就要确保SQL里的字段定义和Model完全一致，否则后面ORM查出来的数据会出现字段错位、类型不匹配的隐性坑。这里我的建议是：开发期用create_all()图快，项目一旦开始涉及字段变更，立刻换成Flask-Migrate。

4. 增删改查四件套：代码实现与细节说明

4.1 新增：session.add与commit的事务逻辑

新增一条数据的标准写法：

from app import db from models import User # 创建一个User对象 user = User(username='zhangsan', email='zhangsan@example.com', age=25) # 加入会话并提交 db.session.add(user) db.session.commit()

这段代码看起来简单，但很多人不理解第一行和第二行的区别。db.session.add(user)只是把对象纳入Session的管理范围，数据库里什么都还没发生。真正把INSERT语句发给MySQL的是db.session.commit()，commit才是事务提交点。

有个常见疑问：不commit会发生什么？如果你只add了，在开发环境下进程继续跑着，数据在Session和事务里确实存在，但你退出程序或者请求结束了，事务回滚，数据就没了。Flask-SQLAlchemy在请求结束时会把Session清掉，未提交的事务默认回滚，所以忘写commit等于白写。

批量新增用add_all：

users = [ User(username='lisi', email='lisi@example.com'), User(username='wangwu', email='wangwu@example.com'), ] db.session.add_all(users) db.session.commit()

还有个实用小技巧：commit之后，自增主键就会回填到对象上。也就是说，db.session.commit()执行完，user.id就变成数据库里真实的那行主键值了，后面想拿id做关联操作很方便。

4.2 查询：filter、filter_by、排序和分页

查询是CRUD里用得最多的部分，也是最容易写成五花八门的部分。基础查询：

# 查所有用户 users = User.query.all() # 查第一条 user = User.query.first() # 按主键查（SQLAlchemy 2.0推荐写法） user = db.session.get(User, 1) # filter: 使用类属性做条件 users = User.query.filter(User.age > 20).all() # filter_by: 使用关键字参数，只能做等值 user = User.query.filter_by(username='zhangsan').first() # 组合条件 users = User.query.filter(User.age > 20, User.is_active == True).all()

filter和filter_by的区别，我用一句话总结：filter更灵活，可以写大于小于不等于各种表达式；filter_by更简洁，但只能做等值判断。新手建议先把filter练熟，因为实际业务条件往往是范围比较、模糊匹配这些，光靠filter_by根本不够。

排序和分页，这是查询里最有含金量的两个操作：

# 按年龄倒序 users = User.query.order_by(User.age.desc()).all() # 分页 page_obj = User.query.paginate(page=1, per_page=10, error_out=False) users = page_obj.items total = page_obj.total page_obj.pages # 总页数

paginate返回的是分页对象，items才是本页的数据列表。前端做分页组件时，这个对象里的total、page、pages都是现成参数。

模糊查询用like：

from sqlalchemy import or_, and_ # 模糊搜索 users = User.query.filter(User.username.like('%zhang%')).all() # or条件 users = User.query.filter(or_(User.username == 'zhangsan', User.email.like('%example.com'))).all()

4.3 更新：单条对象更新与批量update

更新操作有两条路线：对象级更新和查询级批量更新。对象级更新更符合ORM的直觉：

user = db.session.get(User, 1) user.username = 'zhangsan_new' user.age = 26 db.session.commit()

背后的逻辑是：SQLAlchemy会跟踪User这个对象在Session里的属性变化，commit时它自动生成对应的UPDATE语句。你不需要手动写update user set username=...，ORM替你做了。

批量更新则走update()方法，适合一次性改一大批数据的场景：

# 把所有年龄小于18的用户设为active=False updated_count = User.query.filter(User.age < 18).update({'is_active': False}) db.session.commit()

注意这里update()返回的是受影响的行数，而且是一次UPDATE语句完成的，性能上比"循环查出再逐个改"高很多。为什么？循环方案是N条SELECT+N条UPDATE，批量更新是1条UPDATE，差距在数据量大时非常明显。不过批量更新有一个副作用：它不会触发ORM的属性变化跟踪，所以如果要更新模型里定义的updated_at这类自动维护字段，需要在update()的参数里手动带上。

4.4 删除：db.session.delete与批量删除

删除操作也分两种。单条删除：

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

批量删除：

# 删除所有未激活且年龄大于30的用户 deleted_count = User.query.filter(User.is_active == False, User.age > 30).delete() db.session.commit()

删除这块我特别想提"软删除"。实际业务项目里，直接delete物理删数据风险不小：用户误删了找不回来，关联数据会断裂，审计也难做。我在项目里普遍的做法是加一个is_deleted字段，删除操作实际是把这个标记置为True，查询时统一过滤掉。如果需要严格物理删除，再真删。软删除虽然增加一点代码量，但给自己的数据安全兜了底。

注意，软删除的时候查询条件都会多一些，可以用一个BaseModel把公共字段和公共过滤逻辑抽出来，避免每个模型都重复写。

5. 实测中容易踩的坑：从SSL错误到事务回滚

5.1 mysql ssl连接错误与cryptography依赖

搜索热词里就有"mysql ssl连接错误"，这个报错我见得不少。具体报错长这样：

RuntimeError: 'cryptography' package is required for sha256_password or caching_sha2_password auth methods

或者类似：

OperationalError: (pymysql.err.OperationalError) (2059, <cryptography is required ...>)

这里的原因要仔细说明一下。MySQL 8.0默认的caching_sha2_password认证方式，在连接建立时需要做加密的数据交换，这个加密过程依赖cryptography这个库。所以如果只装了pymysql而没装cryptography，连接就被卡住了。

解决办法，先安装依赖：

pip install cryptography

再启动项目，通常就好了。如果还不行，看连接串里是不是带了奇怪的ssl参数。比如用了mysql-connector驱动，有些人会看到ssl_disabled相关的报错，那就是连接串里需要确认是否需要显式禁用SSL：

# 使用mysql-connector驱动时的示例 'mysql+mysqlconnector://root:123456@127.0.0.1:3306/flask_demo?ssl_disabled=true'

排查这类问题，我习惯先分两类：一类是缺包，一类是连接参数不对。前者看报错关键字cryptography，后者看ssl相关关键词。不要一上来就改MySQL配置，那往往不是问题根源。

5.2 中文乱码与charset=utf8mb4

中文乱码是连接MySQL绕不过去的坑。常见报错：

Incorrect string value: '\xE6\x9D\x8E...' for column 'username' at row 1

如果你看到这个错误，说明插入的中文数据没有正确编码。这里的第一个解决点是连接串里写charset=utf8mb4，这个参数会告诉MySQL"我们之间用utf8mb4字符集通信"。注意不是utf8，MySQL的utf8实际上是utf8mb3，连表情符号都存不进去，存emoji会直接报错。

第二个解决点是表本身的字符集。光有连接参数不够，表在创建的时候如果不是utf8mb4，写进去照样乱。建表语句要么显式指定：

CREATE TABLE user ( ... ) DEFAULT CHARSET=utf8mb4;

要么在db.create_all()之前给连接串和模型一个兜底。如果数据库整体是用默认latin1建的，建议在MySQL配置文件里把默认字符集改成utf8mb4，一劳永逸。

实测经验是：只要连接串写对了charset=utf8mb4，新项目基本不会出现中文乱码。乱码多数发生在老库升级、连接串漏参数、表字符集不一致这三个场景。

5.3 事务边界：commit与rollback的时机

事务边界这个问题，我见过太多次"看起来没毛病实际数据丢了"的例子。看这段代码：

@app.route('/user/add', methods=['POST']) def add_user(): user = User(username='zhangsan', email='zhangsan@example.com') db.session.add(user) # 忘了commit return {'status': 'ok'}

接口返回成功了，但数据库里没数据。为什么？因为Flask-SQLAlchemy在请求结束后会清理Session，未commit的事务直接回滚。这种错误不是语法错，也不是运行时报错，是逻辑错，特别隐蔽。

推荐的写法是把事务操作放进try/except里，异常时主动rollback：

@app.route('/user/add', methods=['POST']) def add_user(): try: user = User(username='zhangsan', email='zhangsan@example.com') db.session.add(user) db.session.commit() return {'status': 'ok', 'id': user.id} except Exception as e: db.session.rollback() return {'status': 'error', 'msg': str(e)}

除了commit，还有一个容易误解的点：一个请求里多次commit还是最后commit一次？如果一次请求只涉及一组数据库操作，最好只commit一次。因为每次commit都代表一个事务的结束，多个commit意味着多个事务，一旦中间某个失败，已经提交的前半段是没法回滚的。正确的思路是：一段业务逻辑对应一个事务，事务结束统一commit。

为什么要rollback？commit失败后，事务里可能还留着脏状态，如果不rollback，下一次commit可能会带着这些脏数据继续执行。所以异常分支里rollback不是可选项，是必须项。

6. 从"能跑"到"好用"：ORM实践的一点项目经验

6.1 session的上下文管理与统一提交

到这一节，基本已经能跑通完整的增删改查了。但项目真的变大之后，你会发现在每个视图函数里都写一遍try/except/commit/rollback非常痛苦。代码重复不说，还容易漏。

一个稍微工程化的做法是把增删改查封装成通用方法或者Service层。比如我可以封装一个统一的提交函数：

def save(obj): try: db.session.add(obj) db.session.commit() return obj except Exception: db.session.rollback() raise

然后业务代码里只要：

user = User(username='zhangsan', email='zhangsan@example.com') save(user)

删除、更新也同理。这个封装看似简单，但它的价值在于把"事务边界"这个最容易出错的点收敛到了一个地方。项目里再多人开发，也没人敢在业务代码里乱写commit了。

另外，SQLAlchemy还支持with db.session.begin():这种上下文管理事务的写法：

with db.session.begin(): db.session.add(user1) db.session.add(user2) # 这块代码如果抛异常，自动回滚

这种方式更简洁，适合比较规整的批量操作场景。tornado、FastAPI等其他框架里也有类似的设计思路，学会了到处通用。

6.2 查询常见病：N+1与忽略SQL日志

最后一个经验想聊查询性能。我看到很多新手在循环里查数据库：

for user in users: orders = Order.query.filter_by(user_id=user.id).all() # 处理orders...

这段代码在N个用户时，会执行N条额外的查询，合起来就是N+1条SQL。量小无所谓，但用户量上去之后，接口会肉眼可见地变慢。ORM虽然帮你省了写SQL的功夫，但不会帮你省性能分析。正确的做法是提前用join一次性把关联数据带出来：

results = db.session.query(User, Order).join(Order, User.id == Order.user_id).all()

还有一个调试神器一定要用起来：开启SQLAlchemy的日志输出。在Flask配置里加一行：

app.config['SQLALCHEMY_ECHO'] = True

这个配置会让终端打印出ORM实际执行的每一条SQL语句。排查慢查询、确认事务提交时机、观察批量更新是否真的合并成一条UPDATE，全靠它。建议开发环境一直开着，性能排查和逻辑验证的效率会提高一大截。上线前再关掉就行。

对我来说，ORM从来不是银弹。它解决的是"对象与表结构映射"的繁琐问题，但不是解决所有数据库问题的万能工具。复杂的多表聚合查询、需要精细控制索引的场景，该手写原生SQL就手写。能把ORM和SQL灵活切换使用，才是一个后端开发者真正顺手的状态。

最后再分享一个小技巧。我平时在项目里建一个base_model.py，里面把to_dict()、save()、delete()这类通用操作都写好，然后所有模型继承它。这样视图层代码会干净不少，而且每个新模型都天然拥有一套基础CRUD能力。这套方案在我带了几个项目之后依然觉得实用，推荐你试试。下一篇如果有机会，可以接着聊聊Flask-Migrate和表关系映射，或者说一个完整的用户-订单-商品模型怎么设计。