Python 项目实战：用 Flask 实现 MySQL 数据库增删改查 API

Python 项目实战：用 Flask 实现 MySQL 数据库增删改查 API，本文围绕用 Flask 实现 MySQL 数据库增删改查（CRUD）API 展开，先介绍项目准备，包括 Flask、MySQL、PyMySQL 等技术栈选择，Python 3.6 + 的环境要求，以及 MySQL 数据库和用户表的创建；接着搭建项目结构，编写配置文件存储数据库连接信息；随后分别用 PyMySQL 原生操作和 Flask-SQLAlchemy（ORM 工具）两种方式开发 CRUD API，涵盖数据库连接、新增 / 查询 / 更新 / 删除用户的路由实现与逻辑处理；还讲解了用 Postman 测试各 API 接口的步骤，给出请求方式、URL、请求体及预期响应；最后提供项目优化建议，如参数验证、分页查询等，并附上完整代码示例，帮助开发者掌握 Flask 结合 MySQL 开发 API 的方法。

前言

    Python作为一门简洁、易读、功能强大的编程语言，其基础语法是入门学习的核心。掌握好基础语法，能为后续的编程实践打下坚实的基础。本文将全面讲解Python3的基础语法知识，适合编程初学者系统学习。Python以其简洁优雅的语法和强大的通用性，成为当今最受欢迎的编程语言。本专栏旨在系统性地带你从零基础入门到精通Python核心。无论你是零基础小白还是希望进阶的专业开发者，都将通过清晰的讲解、丰富的实例和实战项目，逐步掌握语法基础、核心数据结构、函数与模块、面向对象编程、文件处理、主流库应用（如数据分析、Web开发、自动化）以及面向对象高级特性，最终具备独立开发能力和解决复杂问题的思维，高效应对数据分析、人工智能、Web应用、自动化脚本等广泛领域的实际需求。

---

🥇 点击进入Python入门专栏，Python凭借简洁易读的语法，是零基础学习编程的理想选择。本专栏专为初学者设计，系统讲解Python核心基础：变量、数据类型、流程控制、函数、文件操作及常用库入门。通过清晰示例与实用小项目，助你快速掌握编程思维，打下坚实根基，迈出自动化办公、数据分析或Web开发的第一步。

🥇 点击进入Python小游戏实战专栏， 寓教于乐，用Python亲手打造经典小游戏！本专栏通过开发贪吃蛇、飞机大战、猜数字、简易版俄罗斯方块等趣味项目，在实践中掌握Python核心语法、面向对象编程、事件处理、图形界面（如Pygame）等关键技能，将枯燥的代码学习转化为可见的成果，让学习编程充满乐趣与成就感，快速提升实战能力。

🥇 点击进入Python小工具实战专栏，告别重复劳动，用Python打造效率神器！本专栏教你开发文件批量处理、自动邮件通知、简易爬虫、桌面提醒、密码生成器、天气查询等实用小工具。聚焦os、shutil、requests、smtplib、schedule等核心库，通过真实场景案例，快速掌握自动化脚本编写技巧，解放双手，显著提升工作与生活效率，让代码真正服务于你的日常。

🥇 点击进入Python爬虫实战专栏，解锁网络数据宝库！本专栏手把手教你使用Python核心库（如requests、BeautifulSoup、Scrapy）构建高效爬虫。从基础网页解析到动态页面抓取、数据存储（CSV/数据库）、反爬策略应对及IP代理使用，通过实战项目（如电商比价、新闻聚合、图片采集、舆情监控），掌握合法合规获取并利用网络数据的核心技能，让数据成为你的超能力。

🥇 点击进入Python项目实战专栏，告别碎片化学习，挑战真实项目！本专栏精选Web应用开发（Flask/Django）、数据分析可视化、自动化办公系统、简易爬虫框架、API接口开发等综合项目。通过需求分析、架构设计、编码实现、测试部署的全流程，深入掌握工程化开发、代码复用、调试排错与团队协作核心能力，积累高质量作品集，真正具备解决复杂问题的Python实战经验。

---

在Web开发中，API常需与数据库交互以实现数据的持久化存储，MySQL作为主流关系型数据库，广泛用于各类项目。本文基于Flask框架，结合PyMySQL库，实现对MySQL数据库的增删改查（CRUD）API，适合有基础Flask知识和MySQL基础的开发者，完整覆盖环境搭建、数据库设计、API开发及测试全流程。

一、项目准备

（一）技术栈选择

• 后端框架：Flask，轻量级Web框架，灵活易扩展，适合快速开发API。
• 数据库：MySQL 8.0及以上版本，稳定可靠的关系型数据库，支持复杂数据查询与事务。
• 数据库连接库：PyMySQL，纯Python实现的MySQL客户端库，用于在Flask项目中连接并操作MySQL数据库。
• 辅助工具：Postman（API测试）、Navicat/MySQL Workbench（MySQL数据库管理）、VS Code/PyCharm（代码编写）。

（二）环境要求

1. Python版本：Python 3.6及以上，确保兼容Flask和PyMySQL的最新特性。
2. 依赖库安装：通过pip命令安装所需库，命令如下：

# 安装Flask
pip install flask
# 安装PyMySQL（连接MySQL）
pip install pymysql
# 安装Flask-SQLAlchemy（可选，ORM工具，简化数据库操作）
pip install flask-sqlalchemy

本文先使用PyMySQL原生操作数据库，再补充Flask-SQLAlchemy的实现方式，满足不同开发习惯需求。

（三）MySQL数据库准备

1. 启动MySQL服务，通过Navicat或MySQL Workbench连接数据库，执行以下SQL语句创建项目所需的数据库和表：

-- 创建数据库（命名为flask_mysql_api）
CREATE DATABASE IF NOT EXISTS flask_mysql_api CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

-- 使用数据库
USE flask_mysql_api;

-- 创建用户表（users），包含id（主键）、name（姓名）、age（年龄）、email（邮箱，唯一）
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age INT NOT NULL CHECK (age > 0),
    email VARCHAR(100) NOT NULL UNIQUE,
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 验证表结构：执行DESCRIBE users;，确认字段名、类型、约束是否正确。

二、项目结构搭建

为保证代码可读性和可维护性，创建如下项目目录结构：

flask-mysql-api/          # 项目根目录
├── app.py                # 主程序（API路由、数据库连接、逻辑处理）
├── config.py             # 配置文件（数据库连接信息、Flask配置）
└── requirements.txt      # 项目依赖清单（可选，方便他人部署）

1. config.py：存储数据库连接配置，避免硬编码，示例代码如下：

# config.py
class Config:
    # MySQL数据库连接信息
    MYSQL_HOST = '127.0.0.1'    # 数据库主机地址（本地为127.0.0.1）
    MYSQL_PORT = 3306           # 数据库端口（默认3306）
    MYSQL_USER = 'root'         # 数据库用户名（根据实际情况修改）
    MYSQL_PASSWORD = '123456'   # 数据库密码（根据实际情况修改）
    MYSQL_DB = 'flask_mysql_api'# 数据库名（与前文创建的一致）
    MYSQL_CHARSET = 'utf8mb4'   # 字符集

# 开发环境配置（继承Config）
class DevelopmentConfig(Config):
    DEBUG = True  # 开启调试模式

# 生产环境配置（继承Config）
class ProductionConfig(Config):
    DEBUG = False  # 关闭调试模式
    # 生产环境可添加数据库连接池配置，提升性能
    MYSQL_POOL_SIZE = 10
    MYSQL_MAX_OVERFLOW = 20

# 配置映射，方便切换环境
config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

2. requirements.txt：记录依赖库及版本，执行pip freeze > requirements.txt生成，内容示例：

Flask==2.3.3
PyMySQL==1.1.0
Flask-SQLAlchemy==3.1.1

三、基于PyMySQL实现CRUD API

（一）数据库连接工具函数

在app.py中，先编写数据库连接和关闭的工具函数，复用连接逻辑，避免重复代码：

# app.py
from flask import Flask, request, jsonify
import pymysql
from config import config

# 初始化Flask应用
app = Flask(__name__)
# 加载配置（默认开发环境）
app.config.from_object(config['default'])

# 数据库连接函数
def get_db_connection():
    try:
        connection = pymysql.connect(
            host=app.config['MYSQL_HOST'],
            port=app.config['MYSQL_PORT'],
            user=app.config['MYSQL_USER'],
            password=app.config['MYSQL_PASSWORD'],
            db=app.config['MYSQL_DB'],
            charset=app.config['MYSQL_CHARSET'],
            cursorclass=pymysql.cursors.DictCursor  # 游标返回字典格式（便于转换为JSON）
        )
        return connection
    except pymysql.Error as e:
        app.logger.error(f"数据库连接失败：{str(e)}")
        return None

# 关闭数据库连接函数
def close_db_connection(connection, cursor):
    if cursor:
        cursor.close()
    if connection:
        connection.close()

（二）实现增删改查API路由

1. 新增用户（Create）

• 请求方式：POST
• 请求URL：/api/users
• 请求体：JSON格式，包含name、age、email字段
• 响应：成功返回新增用户信息（含自增ID），失败返回错误信息

@app.route('/api/users', methods=['POST'])
def create_user():
    # 获取请求体JSON数据
    data = request.get_json()
    # 验证必填字段
    required_fields = ['name', 'age', 'email']
    if not all(field in data for field in required_fields):
        return jsonify({'error': '缺少必填字段（name/age/email）'}), 400

    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        # 执行插入SQL
        sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
        cursor.execute(sql, (data['name'], data['age'], data['email']))
        connection.commit()  # 提交事务

        # 获取新增用户的ID，查询并返回完整信息
        user_id = cursor.lastrowid
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        new_user = cursor.fetchone()

        return jsonify({'message': '用户创建成功', 'user': new_user}), 201  # 201表示创建成功

    except pymysql.IntegrityError as e:
        # 处理唯一约束冲突（如email重复）
        connection.rollback()  # 回滚事务
        return jsonify({'error': '邮箱已存在', 'detail': str(e)}), 409
    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"创建用户失败：{str(e)}")
        return jsonify({'error': '创建用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

2. 查询所有用户（Read All）

• 请求方式：GET
• 请求URL：/api/users
• 响应：返回所有用户列表（JSON格式）

@app.route('/api/users', methods=['GET'])
def get_all_users():
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users ORDER BY create_time DESC")
        users = cursor.fetchall()  # 获取所有数据

        return jsonify({'count': len(users), 'users': users}), 200

    except Exception as e:
        app.logger.error(f"查询所有用户失败：{str(e)}")
        return jsonify({'error': '查询用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

3. 查询单个用户（Read One）

• 请求方式：GET
• 请求URL：/api/users/<int:user_id>（user_id为用户ID）
• 响应：成功返回单个用户信息，失败（如用户不存在）返回错误信息

@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_single_user(user_id):
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        user = cursor.fetchone()  # 获取单条数据

        if not user:
            return jsonify({'error': '用户不存在'}), 404

        return jsonify({'user': user}), 200

    except Exception as e:
        app.logger.error(f"查询用户{user_id}失败：{str(e)}")
        return jsonify({'error': '查询用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

4. 更新用户（Update）

• 请求方式：PUT
• 请求URL：/api/users/<int:user_id>
• 请求体：JSON格式，包含需更新的字段（如name、age、email，至少一个）
• 响应：成功返回更新后的用户信息，失败返回错误信息

@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    # 验证至少有一个可更新字段
    update_fields = ['name', 'age', 'email']
    if not any(field in data for field in update_fields):
        return jsonify({'error': '需提供至少一个更新字段（name/age/email）'}), 400

    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        # 先检查用户是否存在
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用户不存在'}), 404

        # 构建动态更新SQL（避免更新未提供的字段）
        set_clause = ", ".join([f"{field} = %s" for field in data if field in update_fields])
        values = [data[field] for field in data if field in update_fields]
        values.append(user_id)  # 最后添加user_id用于WHERE条件

        sql = f"UPDATE users SET {set_clause} WHERE id = %s"
        cursor.execute(sql, values)
        connection.commit()

        # 查询更新后的用户信息
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        updated_user = cursor.fetchone()

        return jsonify({'message': '用户更新成功', 'user': updated_user}), 200

    except pymysql.IntegrityError as e:
        connection.rollback()
        return jsonify({'error': '邮箱已存在', 'detail': str(e)}), 409
    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"更新用户{user_id}失败：{str(e)}")
        return jsonify({'error': '更新用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

5. 删除用户（Delete）

• 请求方式：DELETE
• 请求URL：/api/users/<int:user_id>
• 响应：成功返回删除确认信息，失败返回错误信息

@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        # 检查用户是否存在
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用户不存在'}), 404

        # 执行删除SQL
        cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
        connection.commit()

        return jsonify({'message': f'用户{user_id}删除成功'}), 200

    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"删除用户{user_id}失败：{str(e)}")
        return jsonify({'error': '删除用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

（三）启动应用

在app.py末尾添加启动代码：

if __name__ == '__main__':
    # 开发环境使用5000端口，生产环境需修改为80/443或其他端口
    app.run(host='0.0.0.0', port=5000, debug=app.config['DEBUG'])

四、基于Flask-SQLAlchemy实现CRUD API（可选）

Flask-SQLAlchemy是Flask的ORM（对象关系映射）扩展，可将Python类映射到数据库表，简化SQL操作。以下是简化版实现：

（一）初始化SQLAlchemy

修改app.py的初始化部分：

# app.py（Flask-SQLAlchemy版）
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from config import config

app = Flask(__name__)
app.config.from_object(config['default'])

# 配置SQLAlchemy连接信息（从config中读取）
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{app.config['MYSQL_USER']}:{app.config['MYSQL_PASSWORD']}@{app.config['MYSQL_HOST']}:{app.config['MYSQL_PORT']}/{app.config['MYSQL_DB']}?charset={app.config['MYSQL_CHARSET']}"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 关闭SQLAlchemy的修改跟踪（提升性能）

# 初始化SQLAlchemy
db = SQLAlchemy(app)

# 定义User模型（映射到users表）
class User(db.Model):
    __tablename__ = 'users'  # 数据库表名
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(50), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    email = db.Column(db.String(100), nullable=False, unique=True)
    create_time = db.Column(db.TIMESTAMP, default=db.func.current_timestamp())

    # 将模型转换为字典（便于返回JSON）
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'age': self.age,
            'email': self.email,
            'create_time': self.create_time.strftime('%Y-%m-%d %H:%M:%S')
        }

（二）简化版CRUD路由示例（以新增和查询为例）

# 新增用户（Flask-SQLAlchemy版）
@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    required_fields = ['name', 'age', 'email']
    if not all(field in data for field in required_fields):
        return jsonify({'error': '缺少必填字段'}), 400

    try:
        # 创建User对象（无需手动写SQL）
        new_user = User(
            name=data['name'],
            age=data['age'],
            email=data['email']
        )
        db.session.add(new_user)  # 添加到会话
        db.session.commit()       # 提交会话（等同于SQL的COMMIT）
        return jsonify({'message': '用户创建成功', 'user': new_user.to_dict()}), 201
    except Exception as e:
       db.session.rollback()  # 回滚事务
        return jsonify({'error': '创建用户失败', 'detail': str(e)}), 500

# 查询所有用户（Flask-SQLAlchemy版）
@app.route('/api/users', methods=['GET'])
def get_all_users():
    try:
        # 等价于SELECT * FROM users ORDER BY create_time DESC
        users = User.query.order_by(User.create_time.desc()).all()
        return jsonify({
            'count': len(users),
            'users': [user.to_dict() for user in users]
        }), 200
    except Exception as e:
        return jsonify({'error': '查询用户失败', 'detail': str(e)}), 500

# 其他路由（查询单个、更新、删除）实现类似，此处省略

五、API测试

完成API开发后，需验证各接口功能是否正常，推荐使用Postman工具，测试步骤如下：

（一）启动应用

运行app.py：

python app.py

控制台显示* Running on http://0.0.0.0:5000/表示启动成功。

（二）测试各接口

1. 新增用户：

   • 方法：POST
   • URL：http://127.0.0.1:5000/api/users
   • 请求体（JSON）：

     {
         "name": "张三",
         "age": 25,
         "email": "zhangsan@example.com"
     }
     

   • 预期响应：状态码201，返回新增用户信息。

2. 查询所有用户：

   • 方法：GET
   • URL：http://127.0.0.1:5000/api/users
   • 预期响应：状态码200，返回包含新增用户的列表。

3. 查询单个用户：

   • 方法：GET
   • URL：http://127.0.0.1:5000/api/users/1（1为新增用户的ID）
   • 预期响应：状态码200，返回ID为1的用户信息。

4. 更新用户：

   • 方法：PUT
   • URL：http://127.0.0.1:5000/api/users/1
   • 请求体（JSON）：

     {
         "age": 26,
         "email": "zhangsan_update@example.com"
     }
     

   • 预期响应：状态码200，返回更新后的用户信息（年龄变为26，邮箱更新）。

5. 删除用户：

   • 方法：DELETE
   • URL：http://127.0.0.1:5000/api/users/1
   • 预期响应：状态码200，返回删除成功的提示信息。

六、项目优化建议

1. 添加请求参数验证：使用marshmallow库对请求数据进行类型、格式验证，提升API健壮性。
2. 实现分页查询：当用户数据量大时，查询所有用户可能导致性能问题，可添加page和per_page参数实现分页。

   # 分页查询示例（PyMySQL版）
   @app.route('/api/users', methods=['GET'])
   def get_all_users():
       page = request.args.get('page', 1, type=int)
       per_page = request.args.get('per_page', 10, type=int)
       offset = (page - 1) * per_page
       
       # 执行带分页的SQL
       cursor.execute("SELECT * FROM users ORDER BY create_time DESC LIMIT %s OFFSET %s", (per_page, offset))
       # ...
   

3. 使用数据库连接池：在生产环境中，频繁创建和关闭数据库连接会消耗资源，可使用DBUtils库实现连接池。
4. 添加日志记录：通过Flask的app.logger记录关键操作和错误信息，便于问题排查。
5. 实现身份认证：对API添加Token验证（如JWT），限制未授权访问，保护数据安全。

七、完整代码

下面是使用PyMySQL实现的完整app.py代码，可直接运行：

from flask import Flask, request, jsonify
import pymysql

# 初始化Flask应用
app = Flask(__name__)
# 加载配置（默认开发环境）

# 数据库连接函数
def get_db_connection():
    try:
        connection = pymysql.connect(
            host="填写数据库ip",
            port=端口,
            user="用户",
            password="密码",
            db="数据库"
        )
        return connection
    except pymysql.Error as e:
        app.logger.error(f"数据库连接失败：{str(e)}")
        return None

# 关闭数据库连接函数
def close_db_connection(connection, cursor):
    if cursor:
        cursor.close()
    if connection:
        connection.close()

# 新增用户
@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    required_fields = ['name', 'age', 'email']
    if not all(field in data for field in required_fields):
        return jsonify({'error': '缺少必填字段（name/age/email）'}), 400

    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
        cursor.execute(sql, (data['name'], data['age'], data['email']))
        connection.commit()

        user_id = cursor.lastrowid
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        new_user = cursor.fetchone()

        return jsonify({'message': '用户创建成功', 'user': new_user}), 201

    except pymysql.IntegrityError as e:
        connection.rollback()
        return jsonify({'error': '邮箱已存在', 'detail': str(e)}), 409
    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"创建用户失败：{str(e)}")
        return jsonify({'error': '创建用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 查询所有用户
@app.route('/api/users', methods=['GET'])
def get_all_users():
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users ORDER BY create_time DESC")
        users = cursor.fetchall()

        return jsonify({'count': len(users), 'users': users}), 200

    except Exception as e:
        app.logger.error(f"查询所有用户失败：{str(e)}")
        return jsonify({'error': '查询用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 查询单个用户
@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_single_user(user_id):
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        user = cursor.fetchone()

        if not user:
            return jsonify({'error': '用户不存在'}), 404

        return jsonify({'user': user}), 200

    except Exception as e:
        app.logger.error(f"查询用户{user_id}失败：{str(e)}")
        return jsonify({'error': '查询用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 更新用户
@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    update_fields = ['name', 'age', 'email']
    if not any(field in data for field in update_fields):
        return jsonify({'error': '需提供至少一个更新字段（name/age/email）'}), 400

    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用户不存在'}), 404

        set_clause = ", ".join([f"{field} = %s" for field in data if field in update_fields])
        values = [data[field] for field in data if field in update_fields]
        values.append(user_id)

        sql = f"UPDATE users SET {set_clause} WHERE id = %s"
        cursor.execute(sql, values)
        connection.commit()

        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        updated_user = cursor.fetchone()

        return jsonify({'message': '用户更新成功', 'user': updated_user}), 200

    except pymysql.IntegrityError as e:
        connection.rollback()
        return jsonify({'error': '邮箱已存在', 'detail': str(e)}), 409
    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"更新用户{user_id}失败：{str(e)}")
        return jsonify({'error': '更新用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 删除用户
@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    connection = None
    cursor = None
    try:
        connection = get_db_connection()
        if not connection:
            return jsonify({'error': '数据库连接失败'}), 500

        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用户不存在'}), 404

        cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
        connection.commit()

        return jsonify({'message': f'用户{user_id}删除成功'}), 200

    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"删除用户{user_id}失败：{str(e)}")
        return jsonify({'error': '删除用户失败', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=app.config['DEBUG'])

地址返回： http://127.0.0.1:5000/api/users

八、总结

本文详细介绍了使用Flask和PyMySQL实现MySQL数据库增删改查API的全过程，包括环境搭建、数据库设计、API开发、测试及优化建议。通过本文的学习，你可以掌握：

1. Flask框架基本使用方法，包括路由定义、请求处理、响应返回。
2. PyMySQL库连接和操作MySQL数据库的技巧，如SQL执行、事务处理、异常捕获。
3. RESTful API设计规范，如HTTP方法与CRUD操作的对应关系、状态码使用。
4. 项目优化思路，提升API的性能、安全性和可维护性。

后续可进一步扩展功能，如添加用户认证、实现更复杂的查询逻辑、集成前端页面等，逐步完善项目。

附录：扩展学习资源

1. 官方资源：

   • Python官网：https://www.python.org
   • PyPI：https://pypi.org（查找第三方库）
   • 安装包等相关文件（另附带pycharm工具），网盘下载地址：https://pan.quark.cn/s/649af731037c
   • 学习资料视频和文档资源，网盘下载地址： https://pan.quark.cn/s/ee16901a8954

2. 本专栏特色资源：

   • 代码资源仓库：CSDN专属资源在线获取
   • 海量Python教程：关注公众号：xcLeigh，获取网盘地址
   • 一对一答疑：添加微信与博主在线沟通（备注“Python专栏”）

3. 相关书籍推荐:

书名	说明
	成功的Python包应该安装简便、运行稳定，并能持续可靠地更新。发布一个完美的Python包需要遵循严谨的流程体系，包括系统化测试和代码审查，以及完善的文档体系。值得庆幸的是，Python生态提供了丰富的工具和技术，可实现从项目初始化到版本发布的全流程自动化。 本书深度解析了以自动化和可扩展的方式共享Python 代码的实用流程。通过实际操作，读者可轻松掌握最新打包工具的运用技巧，深入了解包测试和持续集成的方方面面，甚至可获得创建可持续开源项目的专业技巧，包括许可协议、文档编写及培育贡献者社区等关键维度。🥇 点击购买
	本书涵盖的主题十分广泛，首先介绍贝叶斯推理及其与深度学习的关系。然后探索各种主流的贝叶斯深度学习方法，展示如何在Python和Tensorflow中实现这些方法。本书文笔优美，通俗易懂，即使是不熟悉贝叶斯统计或深度学习的读者也能阅读学习。我特别喜欢第5章。该章很好地解释了PBP和BBB，列举了在Python和Tensorflow中实现和扩展这些方法的示例。强烈建议有兴趣学习BDL的人士阅读本书。我相信，任何读过本书的人都能更深入地理解贝叶斯深度学习。🥇 点击购买

联系博主

    xcLeigh 博主，全栈领域优质创作者，博客专家，目前，活跃在CSDN、微信公众号、小红书、知乎、掘金、快手、思否、微博、51CTO、B站、腾讯云开发者社区、阿里云开发者社区等平台，全网拥有几十万的粉丝，全网统一IP为 xcLeigh。希望通过我的分享，让大家能在喜悦的情况下收获到有用的知识。主要分享编程、开发工具、算法、技术学习心得等内容。很多读者评价他的文章简洁易懂，尤其对于一些复杂的技术话题，他能通过通俗的语言来解释，帮助初学者更好地理解。博客通常也会涉及一些实践经验，项目分享以及解决实际开发中遇到的问题。如果你是开发领域的初学者，或者在学习一些新的编程语言或框架，关注他的文章对你有很大帮助。

    亲爱的朋友，无论前路如何漫长与崎岖，都请怀揣梦想的火种，因为在生活的广袤星空中，总有一颗属于你的璀璨星辰在熠熠生辉，静候你抵达。

     愿你在这纷繁世间，能时常收获微小而确定的幸福，如春日微风轻拂面庞，所有的疲惫与烦恼都能被温柔以待，内心永远充盈着安宁与慰藉。

    至此，文章已至尾声，而您的故事仍在续写，不知您对文中所叙有何独特见解？期待您在心中与我对话，开启思想的新交流。

---

     💞 关注博主 🌀 带你实现畅游前后端！

     🏰 大屏可视化 🌀 带你体验酷炫大屏！

     💯 神秘个人简介 🌀 带你体验不一样得介绍！

     🥇 从零到一学习Python 🌀 带你玩转Python技术流！

     🏆 前沿应用深度测评 🌀 前沿AI产品热门应用在线等你来发掘！

     💦 注：本文撰写于CSDN平台,作者：xcLeigh（所有权归作者所有） ，https://xcleigh.blog.csdn.net/，如果相关下载没有跳转，请查看这个地址，相关链接没有跳转，皆是抄袭本文，转载请备注本文原地址。

---

     📣 亲，码字不易，动动小手，欢迎 点赞 ➕ 收藏，如 🈶 问题请留言（或者关注下方公众号，看见后第一时间回复，还有海量编程资料等你来领！），博主看见后一定及时给您答复 💌💌💌