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

Ne0inhk

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

在这里插入图片描述

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）CREATEDATABASEIFNOTEXISTS flask_mysql_api CHARACTERSET utf8mb4 COLLATE utf8mb4_unicode_ci;-- 使用数据库USE flask_mysql_api;-- 创建用户表（users），包含id（主键）、name（姓名）、age（年龄）、email（邮箱，唯一）CREATETABLEIFNOTEXISTS users ( id INTAUTO_INCREMENTPRIMARYKEY, name VARCHAR(50)NOTNULL, age INTNOTNULLCHECK(age >0), email VARCHAR(100)NOTNULLUNIQUE, create_time TIMESTAMPDEFAULTCURRENT_TIMESTAMP);
  1. 验证表结构：执行DESCRIBE users;，确认字段名、类型、约束是否正确。
在这里插入图片描述

二、项目结构搭建

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

flask-mysql-api/ # 项目根目录 ├── app.py # 主程序（API路由、数据库连接、逻辑处理） ├── config.py # 配置文件（数据库连接信息、Flask配置） └── requirements.txt # 项目依赖清单（可选，方便他人部署）
  1. config.py：存储数据库连接配置，避免硬编码，示例代码如下：
# config.pyclassConfig:# 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）classDevelopmentConfig(Config): DEBUG =True# 开启调试模式# 生产环境配置（继承Config）classProductionConfig(Config): DEBUG =False# 关闭调试模式# 生产环境可添加数据库连接池配置，提升性能 MYSQL_POOL_SIZE =10 MYSQL_MAX_OVERFLOW =20# 配置映射，方便切换环境 config ={'development': DevelopmentConfig,'production': ProductionConfig,'default': DevelopmentConfig }
  1. 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.pyfrom flask import Flask, request, jsonify import pymysql from config import config # 初始化Flask应用 app = Flask(__name__)# 加载配置（默认开发环境） app.config.from_object(config['default'])# 数据库连接函数defget_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)}")returnNone# 关闭数据库连接函数defclose_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'])defcreate_user():# 获取请求体JSON数据 data = request.get_json()# 验证必填字段 required_fields =['name','age','email']ifnotall(field in data for field in required_fields):return jsonify({'error':'缺少必填字段（name/age/email）'}),400 connection =None cursor =Nonetry: connection = get_db_connection()ifnot 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)}),409except Exception as e:if connection: connection.rollback() app.logger.error(f"创建用户失败：{str(e)}")return jsonify({'error':'创建用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)
2. 查询所有用户（Read All）
  • 请求方式：GET
  • 请求URL：/api/users
  • 响应：返回所有用户列表（JSON格式）
@app.route('/api/users', methods=['GET'])defget_all_users(): connection =None cursor =Nonetry: connection = get_db_connection()ifnot 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}),200except Exception as e: app.logger.error(f"查询所有用户失败：{str(e)}")return jsonify({'error':'查询用户失败','detail':str(e)}),500finally: 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'])defget_single_user(user_id): connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor() cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,)) user = cursor.fetchone()# 获取单条数据ifnot user:return jsonify({'error':'用户不存在'}),404return jsonify({'user': user}),200except Exception as e: app.logger.error(f"查询用户{user_id}失败：{str(e)}")return jsonify({'error':'查询用户失败','detail':str(e)}),500finally: 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'])defupdate_user(user_id): data = request.get_json()# 验证至少有一个可更新字段 update_fields =['name','age','email']ifnotany(field in data for field in update_fields):return jsonify({'error':'需提供至少一个更新字段（name/age/email）'}),400 connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor()# 先检查用户是否存在 cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,))ifnot 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}),200except pymysql.IntegrityError as e: connection.rollback()return jsonify({'error':'邮箱已存在','detail':str(e)}),409except Exception as e:if connection: connection.rollback() app.logger.error(f"更新用户{user_id}失败：{str(e)}")return jsonify({'error':'更新用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)
5. 删除用户（Delete）
  • 请求方式：DELETE
  • 请求URL：/api/users/<int:user_id>
  • 响应：成功返回删除确认信息，失败返回错误信息
@app.route('/api/users/<int:user_id>', methods=['DELETE'])defdelete_user(user_id): connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor()# 检查用户是否存在 cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,))ifnot 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}删除成功'}),200except Exception as e:if connection: connection.rollback() app.logger.error(f"删除用户{user_id}失败：{str(e)}")return jsonify({'error':'删除用户失败','detail':str(e)}),500finally: 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表）classUser(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）defto_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'])defcreate_user(): data = request.get_json() required_fields =['name','age','email']ifnotall(field in data for field in required_fields):return jsonify({'error':'缺少必填字段'}),400try:# 创建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()}),201except Exception as e: db.session.rollback()# 回滚事务return jsonify({'error':'创建用户失败','detail':str(e)}),500# 查询所有用户（Flask-SQLAlchemy版）@app.route('/api/users', methods=['GET'])defget_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]}),200except 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
    • 预期响应：状态码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
    • 预期响应：状态码200，返回更新后的用户信息（年龄变为26，邮箱更新）。
  5. 删除用户：
    • 方法：DELETE
    • URL：http://127.0.0.1:5000/api/users/1
    • 预期响应：状态码200，返回删除成功的提示信息。

请求体（JSON）：

{"age":26,"email":"[email protected]"}

请求体（JSON）：

{"name":"张三","age":25,"email":"[email protected]"}

六、项目优化建议

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

实现分页查询：当用户数据量大时，查询所有用户可能导致性能问题，可添加page和per_page参数实现分页。

# 分页查询示例（PyMySQL版）@app.route('/api/users', methods=['GET'])defget_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))# ...

七、完整代码

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

from flask import Flask, request, jsonify import pymysql # 初始化Flask应用 app = Flask(__name__)# 加载配置（默认开发环境）# 数据库连接函数defget_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)}")returnNone# 关闭数据库连接函数defclose_db_connection(connection, cursor):if cursor: cursor.close()if connection: connection.close()# 新增用户@app.route('/api/users', methods=['POST'])defcreate_user(): data = request.get_json() required_fields =['name','age','email']ifnotall(field in data for field in required_fields):return jsonify({'error':'缺少必填字段（name/age/email）'}),400 connection =None cursor =Nonetry: connection = get_db_connection()ifnot 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}),201except pymysql.IntegrityError as e: connection.rollback()return jsonify({'error':'邮箱已存在','detail':str(e)}),409except Exception as e:if connection: connection.rollback() app.logger.error(f"创建用户失败：{str(e)}")return jsonify({'error':'创建用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)# 查询所有用户@app.route('/api/users', methods=['GET'])defget_all_users(): connection =None cursor =Nonetry: connection = get_db_connection()ifnot 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}),200except Exception as e: app.logger.error(f"查询所有用户失败：{str(e)}")return jsonify({'error':'查询用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)# 查询单个用户@app.route('/api/users/<int:user_id>', methods=['GET'])defget_single_user(user_id): connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor() cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,)) user = cursor.fetchone()ifnot user:return jsonify({'error':'用户不存在'}),404return jsonify({'user': user}),200except Exception as e: app.logger.error(f"查询用户{user_id}失败：{str(e)}")return jsonify({'error':'查询用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)# 更新用户@app.route('/api/users/<int:user_id>', methods=['PUT'])defupdate_user(user_id): data = request.get_json() update_fields =['name','age','email']ifnotany(field in data for field in update_fields):return jsonify({'error':'需提供至少一个更新字段（name/age/email）'}),400 connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor() cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,))ifnot 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}),200except pymysql.IntegrityError as e: connection.rollback()return jsonify({'error':'邮箱已存在','detail':str(e)}),409except Exception as e:if connection: connection.rollback() app.logger.error(f"更新用户{user_id}失败：{str(e)}")return jsonify({'error':'更新用户失败','detail':str(e)}),500finally: close_db_connection(connection, cursor)# 删除用户@app.route('/api/users/<int:user_id>', methods=['DELETE'])defdelete_user(user_id): connection =None cursor =Nonetry: connection = get_db_connection()ifnot connection:return jsonify({'error':'数据库连接失败'}),500 cursor = connection.cursor() cursor.execute("SELECT * FROM users WHERE id = %s",(user_id,))ifnot cursor.fetchone():return jsonify({'error':'用户不存在'}),404 cursor.execute("DELETE FROM users WHERE id = %s",(user_id,)) connection.commit()return jsonify({'message':f'用户{user_id}删除成功'}),200except Exception as e:if connection: connection.rollback() app.logger.error(f"删除用户{user_id}失败：{str(e)}")return jsonify({'error':'删除用户失败','detail':str(e)}),500finally: 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. 官方资源：
  2. 本专栏特色资源：
    • 代码资源仓库：ZEEKLOG专属资源在线获取
    • 海量Python教程：关注公众号：xcLeigh，获取网盘地址
    • 一对一答疑：添加微信与博主在线沟通（备注“Python专栏”）
  3. 相关书籍推荐:
书名说明
在这里插入图片描述成功的Python包应该安装简便、运行稳定，并能持续可靠地更新。发布一个完美的Python包需要遵循严谨的流程体系，包括系统化测试和代码审查，以及完善的文档体系。值得庆幸的是，Python生态提供了丰富的工具和技术，可实现从项目初始化到版本发布的全流程自动化。
本书深度解析了以自动化和可扩展的方式共享Python 代码的实用流程。通过实际操作，读者可轻松掌握最新打包工具的运用技巧，深入了解包测试和持续集成的方方面面，甚至可获得创建可持续开源项目的专业技巧，包括许可协议、文档编写及培育贡献者社区等关键维度。🥇 点击购买
在这里插入图片描述本书涵盖的主题十分广泛，首先介绍贝叶斯推理及其与深度学习的关系。然后探索各种主流的贝叶斯深度学习方法，展示如何在Python和Tensorflow中实现这些方法。本书文笔优美，通俗易懂，即使是不熟悉贝叶斯统计或深度学习的读者也能阅读学习。我特别喜欢第5章。该章很好地解释了PBP和BBB，列举了在Python和Tensorflow中实现和扩展这些方法的示例。强烈建议有兴趣学习BDL的人士阅读本书。我相信，任何读过本书的人都能更深入地理解贝叶斯深度学习。🥇 点击购买

联系博主

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

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

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

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

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

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

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

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

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

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

在这里插入图片描述

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

Read more

详解数据结构之跳表

详解数据结构之跳表

目录 跳表的定义 跳表的演化过程 跳表的优化思路 跳表如何保证效率 跳表的时间复杂度 跳表的空间复杂度 跳表的查找 跳表的插入 跳表的删除 跳表的模拟实现 跳表与平衡搜索树及哈希表的对比 跳表的定义 跳表是由William Pugh(音译为威廉·普)发明的，最早出现于他在1990年发表的论文《Skip Lists: A Probabilistic Alternative to Balanced Trees》，跳表全称为跳跃列表，它允许快速查询，插入和删除一个有序连续元素的数据链表。 跳表的演化过程 对于单链表来说，即使数据是已经排好序的，想要查询其中的一个数据，只能从头开始遍历链表，这样效率很低，时间复杂度很高，是 O(n)，如下图所示。 那我们有没有什么办法来提高查询的效率呢？我们可以为链表建立一个“索引”，这样查找起来就会更快，如下图所示，我们在原始链表的基础上，每两个结点提取一个结点建立索引，我们把抽取出来的结点叫作索引层或者索引，down

By Ne0inhk
《算法闯关指南：优选算法-双指针》--03快乐数，04盛水最多的容器

《算法闯关指南：优选算法-双指针》--03快乐数，04盛水最多的容器

🔥草莓熊Lotso：个人主页 ❄️个人专栏：《C++知识分享》《Linux 入门到实践：零基础也能懂》 ✨生活是默默的坚持，毅力是永久的享受。 🎬博主简介： 目录 前言： 03.快乐数 题目分析： 解法：（快慢指针） 算法思路： 补充知识：如何求一个数n每个位置上的数字的平方和 C++代码演示： 算法总结&&笔记展示： 04.盛水最多的容器 解法：（对撞指针） 算法思路： C++代码演示： 算法总结&&笔记展示： 前言： 聚焦算法题实战，系统讲解三大核心板块：优选算法：剖析动态规划、二分法等高效策略，学会寻找“最优解”。 递归与回溯：掌握问题分解与状态回退，攻克组合、排列等难题。

By Ne0inhk
Flutter 三方库 collection 的鸿蒙化适配指南 - 实现具备高级集合操作与相等性深度判定算法的算法底座、支持端侧多维数据结构的高性能治理实战

Flutter 三方库 collection 的鸿蒙化适配指南 - 实现具备高级集合操作与相等性深度判定算法的算法底座、支持端侧多维数据结构的高性能治理实战

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 三方库 collection 的鸿蒙化适配指南 - 实现具备高级集合操作与相等性深度判定算法的算法底座、支持端侧多维数据结构的高性能治理实战 前言 在进行 Flutter for OpenHarmony 开发时，面对复杂的业务 JSON 转化、深层嵌套的集合对比或需要对列表执行高频的优先级排序（Priority Queue）时，原生 List 和 Map 的功能往往显得捉襟见肘。collection 是 Dart 官方维护的最权威、最核心的集合工具库。本文将探讨如何在鸿蒙端构建极致、稳健的数据处理架构。 一、原直观解析 / 概念介绍 1.1 基础原理 该库扩展了 Dart 标准库中的集合能力。它不仅提供了如 Equality（深度相等判定）、PriorityQueue（

By Ne0inhk
【C++】哈希扩展——位图和布隆过滤器的介绍与实现

【C++】哈希扩展——位图和布隆过滤器的介绍与实现

各位读者大佬好，我是落羽！一个坚持不断学习进步的学生。 如果您觉得我的文章还不错，欢迎多多互三分享交流，一起学习进步！ 也欢迎关注我的blog主页:落羽的落羽 文章目录 * 一、位图 * 1. 概念与实现 * 2. std::bitset * 二、布隆过滤器 * 1. 概念 * 2. 布隆过滤器误判率数学推导 * 3. 实现 一、位图 1. 概念与实现 在许多公司的面试题中会考到这样的场景：给40亿个不重复无符号整数，如何快速判断一个数是否在这40亿数中。 如果使用常规思路，每次查询暴力遍历O(N)太慢，排序+二分查找O(NlogN)+O(logN)，内存不足以放下这些数据。 数据是否在给定的整型数据中，结果是在或不在，正好是两种状态，那么可以用一个二进制比特位来代表数据是否存在的信息，比特位为1代表存在，比特位为0代表不在。那么，我们可以设计一个用比特位表示数据是否存在的数据结构——位图！

By Ne0inhk