Flask与MySQL数据库连接实战：从配置到CRUD操作-开发者社区

1. Flask与MySQL数据库连接实战指南

作为Python轻量级Web框架的佼佼者，Flask在数据库操作方面提供了极大的灵活性。不同于直接使用pymysql等驱动编写原生SQL，采用ORM（对象关系映射）技术能让我们的代码更加Pythonic，也更容易维护。本文将手把手带你实现Flask与MySQL的完整对接流程，从环境配置到CRUD操作全覆盖。

我最近在开发一个用户管理系统时，就采用了Flask-SQLAlchemy这套方案。实际体验下来，ORM不仅减少了约60%的数据库相关代码量，还完美解决了SQL注入风险。特别是在团队协作时，统一的ORM接口让代码可读性大幅提升。

2. 环境准备与数据库配置

2.1 安装必要依赖

在开始前，确保已安装Python 3.6+和MySQL 5.7+。然后通过pip安装核心包：

pip install flask flask-sqlalchemy pymysql

这里有个容易踩的坑：如果只安装flask-sqlalchemy而不显式安装pymysql，在某些环境下会出现驱动找不到的错误。我建议总是明确指定这两个依赖。

2.2 创建MySQL数据库

登录MySQL控制台执行：

CREATE DATABASE flask_demo DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

选择utf8mb4字符集非常重要，它能完整支持emoji和所有Unicode字符。曾经有个项目就因为用了utf8导致用户昵称中的特殊符号变成问号，不得不做数据迁移。

3. Flask应用配置

3.1 基础连接配置

新建app.py文件，配置数据库连接：

from flask import Flask from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) # 配置数据库URI app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/flask_demo?charset=utf8mb4' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 关闭警告信息 db = SQLAlchemy(app)

重要提示：生产环境务必不要将密码硬编码在代码中！应该使用环境变量或配置中心管理敏感信息。

3.2 连接测试

添加以下代码验证连接是否成功：

@app.route('/health') def health_check(): try: db.session.execute('SELECT 1') return 'Database connection OK' except Exception as e: return f'Connection failed: {str(e)}', 500

启动应用后访问/health端点，看到"Database connection OK"即表示配置正确。

4. 模型定义与表操作

4.1 创建用户模型

class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) password = db.Column(db.String(120), nullable=False) created_at = db.Column(db.DateTime, server_default=db.func.now()) def __repr__(self): return f'<User {self.username}>'

模型设计要点：

显式指定__tablename__避免表名猜测
添加created_at字段记录创建时间
为username添加unique约束防止重复
实现__repr__方便调试

4.2 初始化数据库表

在Flask shell中执行：

from app import db db.create_all()

或者添加CLI命令：

@app.cli.command() def initdb(): """Initialize the database.""" db.create_all() print('Initialized the database.')

之后通过flask initdb命令即可初始化表结构。

5. CRUD操作实战

5.1 创建用户

from datetime import datetime new_user = User( username='john_doe', password='secure_password', # 实际项目应该存储hash值 created_at=datetime.utcnow() ) db.session.add(new_user) db.session.commit()

安全提醒：实际项目中密码必须使用bcrypt等库进行哈希处理，绝对不要明文存储！

5.2 查询操作

基本查询：

# 获取所有用户 users = User.query.all() # 按条件查询 user = User.query.filter_by(username='john_doe').first() # 复杂查询 from sqlalchemy import or_ admins = User.query.filter( or_( User.username.like('%admin%'), User.id.in_([1, 2, 3]) ) ).order_by(User.created_at.desc()).limit(10).all()

5.3 更新操作

user = User.query.get(1) # 获取ID为1的用户 if user: user.password = 'new_hashed_password' db.session.commit()

批量更新：

User.query.filter(User.id > 10).update({'password': 'updated'}) db.session.commit()

5.4 删除操作

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

6. 高级技巧与性能优化

6.1 连接池配置

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_size': 10, 'max_overflow': 20, 'pool_timeout': 30, 'pool_recycle': 3600 }

合理配置连接池可以显著提升高并发下的性能。根据我的经验，pool_size设为CPU核心数的2-3倍是个不错的起点。

6.2 事务管理

try: # 操作1 user1 = User(username='user1') db.session.add(user1) # 操作2 user2 = User(username='user2') db.session.add(user2) db.session.commit() except Exception as e: db.session.rollback() print(f"Transaction failed: {e}")

6.3 性能监控

添加以下配置记录慢查询：

app.config['SQLALCHEMY_RECORD_QUERIES'] = True from flask_sqlalchemy import get_debug_queries @app.after_request def after_request(response): for query in get_debug_queries(): if query.duration >= 0.5: # 超过500ms的查询 print(f'SLOW QUERY: {query.statement} {query.parameters} {query.duration}') return response

7. 常见问题排查

7.1 连接超时问题

症状：间歇性出现"MySQL server has gone away"错误。

解决方案：

检查MySQL的wait_timeout设置（默认8小时）
配置pool_recycle小于wait_timeout
实现连接健康检查

7.2 编码问题

确保所有环节统一使用utf8mb4：

数据库创建时指定
连接字符串添加charset参数
Python文件头部添加编码声明

7.3 会话管理

常见错误：在请求结束后尝试使用数据库会话。

正确做法：

@app.teardown_appcontext def shutdown_session(exception=None): db.session.remove()

8. 安全最佳实践

使用参数化查询防止SQL注入（ORM已自动处理）
密码必须加盐哈希存储（推荐使用Flask-Bcrypt）
生产环境禁用DEBUG模式
定期备份数据库
为数据库用户分配最小必要权限

我在实际项目中总结出一个有用的技巧：为每个模型添加通用的审计字段：

class BaseModel(db.Model): __abstract__ = True created_at = db.Column(db.DateTime, server_default=db.func.now()) updated_at = db.Column(db.DateTime, onupdate=db.func.now()) is_deleted = db.Column(db.Boolean, default=False) class User(BaseModel): # 其他字段...

这种设计便于实现软删除和操作追踪。