Flask连接MySQL数据库配置教程详解步骤与常见问题解决

引言
Flask是一个轻量级的Python Web框架，而MySQL是世界上最流行的开源关系型数据库之一。在实际的Web开发中，将Flask与MySQL结合使用是非常常见的需求。本文将详细介绍如何在Flask中配置和连接MySQL数据库，包括完整的步骤、代码示例以及常见问题的解决方案。
1. 环境准备
在开始配置之前，我们需要确保系统中已经安装了必要的软件和库。
1.1 安装Python和Flask
确保你的系统中已经安装了Python（推荐3.6+版本）和Flask。如果没有安装，可以通过以下命令安装：
# 安装Flask
pip install flask

# 或者使用虚拟环境（推荐）
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
pip install flask
1.2 安装MySQL数据库
你需要在本地或远程服务器上安装MySQL数据库。可以从
MySQL官方网站
下载并安装。
1.3 安装必要的Python库
Flask本身不直接支持MySQL，我们需要安装额外的库来连接MySQL。常用的有以下几种选择：
mysql-connector-python
：Oracle官方提供的纯Python驱动
PyMySQL
：纯Python实现的MySQL客户端
mysqlclient
：MySQLdb的兼容版本，性能较好
# 推荐安装mysql-connector-python
pip install mysql-connector-python

# 或者安装PyMySQL
pip install pymysql

# 或者安装mysqlclient（可能需要系统依赖）
pip install mysqlclient
2. 基本配置步骤
2.1 创建Flask应用和数据库配置
首先，我们创建一个基本的Flask应用，并配置数据库连接信息。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os

app = Flask(__name__)

# 数据库配置
# 格式：mysql://用户名:密码@主机:端口/数据库名
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘mysql://root:password@localhost:3306/mydatabase’
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False

# 初始化数据库扩展
db = SQLAlchemy(app)

# 测试路由
@app.route(‘/’)
def index():
return “Flask MySQL连接成功！”

if __name__ == ‘__main__’:
app.run(debug=True)
2.2 使用环境变量存储敏感信息
在实际项目中，我们不应该将数据库密码等敏感信息直接写在代码中。推荐使用环境变量：
import os
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 从环境变量获取配置
db_user = os.getenv(‘DB_USER’, ‘root’)
db_password = os.getenv(‘DB_PASSWORD’, ‘password’)
db_host = os.getenv(‘DB_HOST’, ‘localhost’)
db_port = os.getenv(‘DB_PORT’, ‘3306’)
db_name = os.getenv(‘DB_NAME’, ‘mydatabase’)

app.config[‘SQLALCHEMY_DATABASE_URI’] = f’mysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}’
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False

db = SQLAlchemy(app)
2.3 创建数据库模型
使用SQLAlchemy ORM创建数据库模型：
from datetime import datetime

class User(db.Model):
__tablename__ = ‘users’

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)
password_hash = db.Column(db.String(128))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

def __repr__(self):
return f’‘

# 创建所有表
with app.app_context():
db.create_all()
3. 数据库操作示例
3.1 增删改查操作
from flask import request, jsonify

@app.route(‘/user/add’, methods=[‘POST’])
def add_user():
data = request.get_json()
username = data.get(‘username’)
email = data.get(’email’)
password = data.get(‘password’)

# 检查用户是否已存在
existing_user = User.query.filter_by(username=username).first()
if existing_user:
return jsonify({‘error’: ‘用户名已存在’}), 400

# 创建新用户
new_user = User(
username=username,
email=email,
password_hash=password # 实际应用中应该加密
)

try:
db.session.add(new_user)
db.session.commit()
return jsonify({
‘message’: ‘用户创建成功’,
‘user_id’: new_user.id
}), 201
except Exception as e:
db.session.rollback()
return jsonify({‘error’: str(e)}), 500

@app.route(‘/user/‘, methods=[‘GET’])
def get_user(user_id):
user = User.query.get(user_id)
if user:
return jsonify({
‘id’: user.id,
‘username’: user.username,
’email’: user.email,
‘created_at’: user.created_at.isoformat()
})
return jsonify({‘error’: ‘用户不存在’}), 404

@app.route(‘/user/‘, methods=[‘PUT’])
def update_user(user_id):
user = User.query.get(user_id)
if not user:
return jsonify({‘error’: ‘用户不存在’}), 404

data = request.get_json()
if ’email’ in data:
user.email = data[’email’]

try:
db.session.commit()
return jsonify({‘message’: ‘用户更新成功’})
except Exception as e:
db.session.rollback()
return jsonify({‘error’: str(e)}), 500

@app.route(‘/user/‘, methods=[‘DELETE’])
def delete_user(user_id):
user = User.query.get(user_id)
if not user:
return jsonify({‘error’: ‘用户不存在’}), 404

try:
db.session.delete(user)
db.session.commit()
return jsonify({‘message’: ‘用户删除成功’})
except Exception as e:
db.session.rollback()
return jsonify({‘error’: str(e)}), 500
3.2 复杂查询示例
@app.route(‘/users/search’)
def search_users():
# 获取查询参数
username = request.args.get(‘username’, ”)
email = request.args.get(’email’, ”)
page = int(request.args.get(‘page’, 1))
per_page = int(request.args.get(‘per_page’, 10))

# 构建查询
query = User.query

if username:
query = query.filter(User.username.like(f’%{username}%’))
if email:
query = query.filter(User.email.like(f’%{email}%’))

# 分页查询
users = query.order_by(User.created_at.desc()).paginate(
page=page,
per_page=per_page,
error_out=False
)

return jsonify({
‘users’: [{
‘id’: user.id,
‘username’: user.username,
’email’: user.email,
‘created_at’: user.created_at.isoformat()
} for user in users.items],
‘total’: users.total,
‘pages’: users.pages,
‘current_page’: users.page
})
4. 使用连接池优化性能
在生产环境中，使用连接池可以显著提高数据库性能：
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 配置连接池
engine = create_engine(
‘mysql://root:password@localhost:3306/mydatabase’,
pool_size=10, # 连接池大小
max_overflow=20, # 超出pool_size的连接数
pool_timeout=30, # 获取连接超时时间（秒）
pool_recycle=3600, # 连接回收时间（秒）
echo=False # 是否打印SQL语句（调试时设为True）
)

# 创建会话工厂
db_session = scoped_session(sessionmaker(bind=engine))

# 基类
Base = declarative_base()
Base.query = db_session.query_property()

# 在应用中使用
@app.teardown_appcontext
def shutdown_session(exception=None):
db_session.remove()
5. 常见问题及解决方案
5.1 连接失败问题
问题描述
：
OperationalError: (2003, “Can’t connect to MySQL server on ‘localhost’”)
可能原因及解决方案
：
MySQL服务未启动
：
“`bash
Linux/Mac
sudo systemctl start mysql
或
sudo service mysql start
# Windows
# 在服务管理器中启动MySQL服务
2. **防火墙阻止连接**：
“`bash
# 检查3306端口是否开放
telnet localhost 3306

# 如果是远程连接，确保防火墙允许3306端口
sudo ufw allow 3306/tcp
MySQL用户权限问题
：
“`sql
– 登录MySQL
mysql -u root -p
– 创建用户并授权
CREATE USER ‘flaskuser’@‘localhost’ IDENTIFIED BY ‘password’;
GRANT ALL PRIVILEGES ON mydatabase.* TO ‘flaskuser’@‘localhost’;
FLUSH PRIVILEGES;
### 5.2 认证失败问题

**问题描述**：`OperationalError: (1045, “Access denied for user ‘root’@’localhost’”)`

**解决方案**：

1. **检查用户名和密码**：
“`python
# 确保密码中的特殊字符被正确转义
import urllib.parse
password = urllib.parse.quote_plus(‘your@password#123’)
app.config[‘SQLALCHEMY_DATABASE_URI’] = f’mysql://root:{password}@localhost:3306/mydatabase’
重置MySQL root密码
：
“`bash
停止MySQL服务
sudo systemctl stop mysql
# 以安全模式启动
sudo mysqld_safe –skip-grant-tables &
# 登录并修改密码
mysql -u root
FLUSH PRIVILEGES;
ALTER USER ‘root’@‘localhost’ IDENTIFIED BY ‘newpassword’;
### 5.3 数据库不存在问题

**问题描述**：`OperationalError: (1049, “Unknown database ‘mydatabase’”)`

**解决方案**：

“`python
# 方案1：手动创建数据库
# 在MySQL中执行：
# CREATE DATABASE mydatabase CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

# 方案2：在连接字符串中指定编码
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘mysql://root:password@localhost:3306/mydatabase?charset=utf8mb4′

# 方案3：使用代码自动创建数据库（需要额外权限）
import mysql.connector

def create_database_if_not_exists():
try:
# 连接到MySQL服务器（不指定具体数据库）
conn = mysql.connector.connect(
host=’localhost’,
user=’root’,
password=’password’
)
cursor = conn.cursor()
cursor.execute(“CREATE DATABASE IF NOT EXISTS mydatabase CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci”)
cursor.close()
conn.close()
except Exception as e:
print(f”创建数据库失败: {e}”)

# 在应用启动时调用
with app.app_context():
create_database_if_not_exists()
db.create_all()
5.4 编码问题
问题描述
：插入中文字符时出现乱码或
Incorrect string value
错误。
解决方案
：
# 1. 确保数据库和表使用utf8mb4编码
# 在创建模型时指定：
class User(db.Model):
__tablename__ = ‘users’
__table_args__ = {
‘mysql_charset’: ‘utf8mb4’,
‘mysql_collate’: ‘utf8mb4_unicode_ci’
}

id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), nullable=False)

# 2. 连接字符串中指定编码
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘mysql://root:password@localhost:3306/mydatabase?charset=utf8mb4’

# 3. 在MySQL配置文件中设置默认编码（my.cnf或my.ini）
# [mysqld]
# character-set-server = utf8mb4
# collation-server = utf8mb4_unicode_ci
# [client]
# default-character-set = utf8mb4
5.5 连接超时问题
问题描述
：长时间运行的应用出现连接超时错误。
解决方案
：
# 1. 使用连接池并设置回收时间
engine = create_engine(
‘mysql://root:password@localhost:3306/mydatabase’,
pool_recycle=3600, # 每小时回收连接
pool_pre_ping=True # 在获取连接时检查连接是否有效
)

# 2. 在Flask配置中设置SQLAlchemy选项
app.config[‘SQLALCHEMY_ENGINE_OPTIONS’] = {
‘pool_recycle’: 3600,
‘pool_pre_ping’: True
}

# 3. 使用try-except处理连接错误
@app.route(‘/test’)
def test_connection():
try:
# 执行一个简单查询
db.session.execute(‘SELECT 1’)
return “连接正常”
except Exception as e:
# 重新初始化连接
db.session.rollback()
return f”连接错误: {e}”, 500
5.6 性能优化问题
问题描述
：查询速度慢，数据库负载高。
解决方案
：
# 1. 使用索引
class User(db.Model):
__tablename__ = ‘users’

id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, index=True) # 添加索引
email = db.Column(db.String(120), unique=True, index=True)
created_at = db.Column(db.DateTime, index=True)

# 2. 使用延迟加载和急切加载
from sqlalchemy.orm import joinedload

# 坏的查询（N+1问题）
users = User.query.all()
for user in users:
print(user.posts) # 每次访问都会查询数据库

# 好的查询（使用joinedload）
from sqlalchemy.orm import joinedload
users = User.query.options(joinedload(User.posts)).all()

# 3. 使用数据库连接池
app.config[‘SQLALCHEMY_POOL_SIZE’] = 20
app.config[‘SQLALCHEMY_POOL_RECYCLE’] = 3600
app.config[‘SQLALCHEMY_POOL_TIMEOUT’] = 30

# 4. 查询优化
# 使用聚合查询而不是循环查询
from sqlalchemy import func

# 坏的做法
user_count = 0
for user in User.query.all():
user_count += 1

# 好的做法
user_count = db.session.query(func.count(User.id)).scalar()
6. 生产环境最佳实践
6.1 使用配置文件管理
# config.py
import os

class Config:
SECRET_KEY = os.environ.get(‘SECRET_KEY’) or ‘dev-secret-key’
SQLALCHEMY_DATABASE_URI = os.environ.get(‘DATABASE_URL’) or \
‘mysql://root:password@localhost:3306/mydatabase’
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_POOL_SIZE = 20
SQLALCHEMY_POOL_RECYCLE = 3600

class DevelopmentConfig(Config):
DEBUG = True

class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_POOL_SIZE = 10
SQLALCHEMY_POOL_RECYCLE = 1800

# app.py
from flask import Flask
from config import DevelopmentConfig

app = Flask(__name__)
app.config.from_object(DevelopmentConfig)

# 数据库初始化
db = SQLAlchemy(app)
6.2 使用数据库迁移
# 安装Flask-Migrate
pip install flask-migrate

# 初始化
flask db init

# 创建迁移脚本
flask db migrate -m “Initial migration”

# 应用迁移
flask db upgrade
# app.py
from flask_migrate import Migrate

migrate = Migrate(app, db)
6.3 错误处理和日志记录
import logging
from logging.handlers import RotatingFileHandler

# 配置日志
handler = RotatingFileHandler(‘app.log’, maxBytes=10000, backupCount=3)
handler.setLevel(logging.INFO)
app.logger.addHandler(handler)

@app.errorhandler(500)
def internal_error(error):
app.logger.error(f”数据库错误: {error}”)
return jsonify({‘error’: ‘服务器内部错误’}), 500

@app.errorhandler(404)
def not_found(error):
return jsonify({‘error’: ‘资源不存在’}), 404
7. 完整项目结构示例
myflaskapp/
├── app/
│ ├── __init__.py
│ ├── models.py
│ ├── routes.py
│ └── config.py
├── migrations/
├── requirements.txt
├── run.py
└── .env
app/
init
.py
:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from config import Config

db = SQLAlchemy()
migrate = Migrate()

def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)

db.init_app(app)
migrate.init_app(app, db)

from app.routes import bp
app.register_blueprint(bp)

return app
run.py
:
from app import create_app

app = create_app()

if __name__ == ‘__main__’:
app.run()
8. 测试数据库连接
# test_connection.py
def test_db_connection():
try:
from app import create_app
app = create_app()

with app.app_context():
# 测试简单查询
db.session.execute(‘SELECT 1’)
print(“✅ 数据库连接成功！”)
return True
except Exception as e:
print(f”❌ 数据库连接失败: {e}”)
return False

if __name__ == ‘__main__’:
test_db_connection()
9. 总结
本文详细介绍了Flask连接MySQL数据库的完整流程，包括：
环境准备
：安装必要的库和工具
基本配置
：创建Flask应用和数据库连接
数据库操作
：增删改查和复杂查询示例
性能优化
：连接池、索引、查询优化
常见问题
：连接失败、编码问题、超时问题等
生产环境最佳实践
：配置管理、数据库迁移、错误处理
通过遵循这些步骤和最佳实践，你可以构建一个稳定、高效的Flask-MySQL应用。记住在生产环境中始终使用环境变量管理敏感信息，并实施适当的错误处理和日志记录机制。
# Flask连接MySQL数据库配置教程详解步骤与常见问题解决
引言
Flask是一个轻量级的Python Web框架，而MySQL是世界上最流行的开源关系型数据库之一。在实际的Web开发中，将Flask与MySQL结合使用是非常常见的需求。本文将详细介绍如何在Flask中配置和连接MySQL数据库，包括完整的步骤、代码示例以及常见问题的解决方案。
1. 环境准备
在开始配置之前，我们需要确保系统中已经安装了必要的软件和库。
1.1 安装Python和Flask
确保你的系统中已经安装了Python（推荐3.6+版本）和Flask。如果没有安装，可以通过以下命令安装：
# 安装Flask
pip install flask