在现代Web应用开发中，数据持久化是一个必不可少的部分，Flask作为一个轻量级的Python Web框架，提供了简洁的接口来与各种数据库进行交互，本文将详细介绍如何使用Flask连接MySQL数据库，并通过SQLAlchemy ORM（对象关系映射）工具简化数据库操作。

一、准备工作

1. 安装必要的库

在使用Flask连接MySQL之前，需要确保安装了以下Python库：

- Flask: 轻量级Web框架。

- Flask-SQLAlchemy: SQLAlchemy for Flask的集成。

- mysqlclient: MySQL数据库连接器。

- pymysql: 另一种MySQL数据库连接器（可选）。

使用pip安装上述库：

pip install Flask
pip install flask_sqlalchemy
pip install mysqlclient

如果更喜欢使用pymysql，可以通过以下命令安装：

pip install pymysql

2. 配置MySQL数据库

确保MySQL数据库已经安装并正在运行，创建一个新数据库用于Flask应用：

CREATE DATABASE flask_db CHARACTER SET utf8mb4;

二、创建Flask应用并配置数据库连接

1. 初始化Flask应用

创建一个新的Flask应用，在项目目录下创建一个名为app.py的文件，初始化Flask应用：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)

2. 配置数据库URI

在Flask应用中配置数据库连接信息，这些配置会放在配置文件中，但在这里我们直接在代码中配置：

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqlconnector://username:password@localhost/flask_db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

请根据实际情况替换username和password为你的MySQL数据库用户名和密码，如果使用pymysql，则URI格式如下：

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost/flask_db'

3. 初始化SQLAlchemy

在Flask应用中初始化SQLAlchemy：

db = SQLAlchemy(app)

三、定义数据模型

使用SQLAlchemy ORM定义数据模型，我们定义一个用户模型：

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 f'<User {self.username}>'

在这个例子中，我们定义了一个User类，包含id、username和email三个字段。id是主键，username和email是唯一且不能为空的。

1. 创建数据库表

在定义了模型之后，需要创建数据库表，可以使用SQLAlchemy的create_all方法：

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

四、进行数据库操作

1. 插入数据

向数据库中插入一条新的用户记录：

new_user = User(username='john_doe', email='john@example.com')
db.session.add(new_user)
db.session.commit()
print(f'Added: {new_user}')

2. 查询数据

查询所有用户：

users = User.query.all()
for user in users:
    print(user)

根据用户名查询用户：

user = User.query.filter_by(username='john_doe').first()
print(user.email)

3. 更新数据

更新用户邮箱：

user = User.query.filter_by(username='john_doe').first()
user.email = 'john.doe@example.com'
db.session.commit()
print(f'Updated: {user}')

4. 删除数据

删除用户：

user = User.query.filter_by(username='john_doe').first()
db.session.delete(user)
db.session.commit()
print('Deleted user')

五、最佳实践和常见问题

1. 使用环境变量管理配置

为了安全起见，不要将数据库连接信息硬编码到代码中，可以使用环境变量管理这些敏感信息：

import os
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL')

并在系统环境变量或.env文件中设置DATABASE_URL。

2. 异常处理

在进行数据库操作时，务必添加异常处理，以确保应用的稳定性：

try:
    db.session.commit()
except Exception as e:
    db.session.rollback()
    print(f"Error: {e}")

3. 数据库迁移

在开发过程中，数据库模式可能会发生变化，使用Alembic等数据库迁移工具可以简化这个过程：

pip install alembic
alembic init alembic

然后在alembic.ini中配置环境和元数据：

sqlalchemy.url = driver://user:pass@localhost/dbname

初始化Alembic后，使用以下命令生成和应用迁移脚本：

alembic revision --autogenerate -m "initial migration"
alembic upgrade head

六、实际应用案例

1. 用户认证系统

一个简单的用户认证系统通常需要进行用户数据的存储和检索，以下是一个完整的示例：

from flask import Flask, request, jsonify
from werkzeug.security import generate_password_hash, check_password_hash
from flask_sqlalchemy import SQLAlchemy
from flask import Flask, request, jsonify, abort, make_response
from functools import wraps
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqlconnector://username:password@localhost/flask_db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(128), nullable=False)
    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)
    
    def __repr__(self):
        return f'<User {self.username}>'
创建数据库表
with app.app_context():
    db.create_all()
注册装饰器
def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('x-access-token')
        if not token:
            return jsonify({'message': 'Token is missing!'}), abort(401)
        user = User.verify_auth_token(token)
        if not user:
            return jsonify({'message': 'Token is invalid or expired!'}), abort(401)
        return f(user, *args, **kwargs)
    return decorated
@app.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    hashed_password = generate_password_hash(data['password'], method='sha256')
    new_user = User(username=data['username'], password_hash=hashed_password)
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'message': 'User created!'})
@app.route('/login', methods=['POST'])
def login():
    auth = request.authorization
    if not auth or not