Python用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫(kù)的增刪改查API
更新時(shí)間：2025年09月13日 14:18:42 作者：xcLeigh
Web開(kāi)發(fā)中,API常需與數(shù)據(jù)庫(kù)交互以實(shí)現(xiàn)數(shù)據(jù)的持久化存儲(chǔ),MySQL作為主流關(guān)系型數(shù)據(jù)庫(kù),廣泛用于各類項(xiàng)目,本文基于Flask框架,結(jié)合PyMySQL庫(kù),實(shí)現(xiàn)對(duì)MySQL數(shù)據(jù)庫(kù)的增刪改查（CRUD）API,適合有基礎(chǔ)Flask知識(shí)和MySQL基礎(chǔ)的開(kāi)發(fā)者,完整覆蓋環(huán)境搭建、數(shù)據(jù)庫(kù)設(shè)計(jì)、API開(kāi)發(fā)及測(cè)試
Python作為一門簡(jiǎn)潔、易讀、功能強(qiáng)大的編程語(yǔ)言，其基礎(chǔ)語(yǔ)法是入門學(xué)習(xí)的核心。掌握好基礎(chǔ)語(yǔ)法，能為后續(xù)的編程實(shí)踐打下堅(jiān)實(shí)的基礎(chǔ)。本文將全面講解Python3的基礎(chǔ)語(yǔ)法知識(shí)，適合編程初學(xué)者系統(tǒng)學(xué)習(xí)。Python以其簡(jiǎn)潔優(yōu)雅的語(yǔ)法和強(qiáng)大的通用性，成為當(dāng)今最受歡迎的編程語(yǔ)言。本專欄旨在系統(tǒng)性地帶你從零基礎(chǔ)入門到精通Python核心。無(wú)論你是零基礎(chǔ)小白還是希望進(jìn)階的專業(yè)開(kāi)發(fā)者，都將通過(guò)清晰的講解、豐富的實(shí)例和實(shí)戰(zhàn)項(xiàng)目，逐步掌握語(yǔ)法基礎(chǔ)、核心數(shù)據(jù)結(jié)構(gòu)、函數(shù)與模塊、面向?qū)ο缶幊獭⑽募幚?、主流?kù)應(yīng)用（如數(shù)據(jù)分析、Web開(kāi)發(fā)、自動(dòng)化）以及面向?qū)ο蟾呒?jí)特性，最終具備獨(dú)立開(kāi)發(fā)能力和解決復(fù)雜問(wèn)題的思維，高效應(yīng)對(duì)數(shù)據(jù)分析、人工智能、Web應(yīng)用、自動(dòng)化腳本等廣泛領(lǐng)域的實(shí)際需求。
在Web開(kāi)發(fā)中，API常需與數(shù)據(jù)庫(kù)交互以實(shí)現(xiàn)數(shù)據(jù)的持久化存儲(chǔ)，MySQL作為主流關(guān)系型數(shù)據(jù)庫(kù)，廣泛用于各類項(xiàng)目。本文基于Flask框架，結(jié)合PyMySQL庫(kù)，實(shí)現(xiàn)對(duì)MySQL數(shù)據(jù)庫(kù)的增刪改查（CRUD）API，適合有基礎(chǔ)Flask知識(shí)和MySQL基礎(chǔ)的開(kāi)發(fā)者，完整覆蓋環(huán)境搭建、數(shù)據(jù)庫(kù)設(shè)計(jì)、API開(kāi)發(fā)及測(cè)試全流程。
一、項(xiàng)目準(zhǔn)備
（一）技術(shù)棧選擇
- 后端框架：Flask，輕量級(jí)Web框架，靈活易擴(kuò)展，適合快速開(kāi)發(fā)API。
- 數(shù)據(jù)庫(kù)：MySQL 8.0及以上版本，穩(wěn)定可靠的關(guān)系型數(shù)據(jù)庫(kù)，支持復(fù)雜數(shù)據(jù)查詢與事務(wù)。
- 數(shù)據(jù)庫(kù)連接庫(kù)：PyMySQL，純Python實(shí)現(xiàn)的MySQL客戶端庫(kù)，用于在Flask項(xiàng)目中連接并操作MySQL數(shù)據(jù)庫(kù)。
- 輔助工具：Postman（API測(cè)試）、Navicat/MySQL Workbench（MySQL數(shù)據(jù)庫(kù)管理）、VS Code/PyCharm（代碼編寫）。
（二）環(huán)境要求
- Python版本：Python 3.6及以上，確保兼容Flask和PyMySQL的最新特性。
- 依賴庫(kù)安裝：通過(guò)pip命令安裝所需庫(kù)，命令如下：
# 安裝Flask pip install flask # 安裝PyMySQL（連接MySQL） pip install pymysql # 安裝Flask-SQLAlchemy（可選，ORM工具，簡(jiǎn)化數(shù)據(jù)庫(kù)操作） pip install flask-sqlalchemy
本文先使用PyMySQL原生操作數(shù)據(jù)庫(kù)，再補(bǔ)充Flask-SQLAlchemy的實(shí)現(xiàn)方式，滿足不同開(kāi)發(fā)習(xí)慣需求。
（三）MySQL數(shù)據(jù)庫(kù)準(zhǔn)備
- 啟動(dòng)MySQL服務(wù)，通過(guò)Navicat或MySQL Workbench連接數(shù)據(jù)庫(kù)，執(zhí)行以下SQL語(yǔ)句創(chuàng)建項(xiàng)目所需的數(shù)據(jù)庫(kù)和表：
-- 創(chuàng)建數(shù)據(jù)庫(kù)（命名為flask_mysql_api）
CREATE DATABASE IF NOT EXISTS flask_mysql_api CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
-- 使用數(shù)據(jù)庫(kù)
USE flask_mysql_api;
-- 創(chuàng)建用戶表（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
);
- 驗(yàn)證表結(jié)構(gòu)：執(zhí)行
DESCRIBE users;，確認(rèn)字段名、類型、約束是否正確。
二、項(xiàng)目結(jié)構(gòu)搭建
為保證代碼可讀性和可維護(hù)性，創(chuàng)建如下項(xiàng)目目錄結(jié)構(gòu)：
flask-mysql-api/ # 項(xiàng)目根目錄 ├── app.py # 主程序（API路由、數(shù)據(jù)庫(kù)連接、邏輯處理） ├── config.py # 配置文件（數(shù)據(jù)庫(kù)連接信息、Flask配置） └── requirements.txt # 項(xiàng)目依賴清單（可選，方便他人部署）
- config.py：存儲(chǔ)數(shù)據(jù)庫(kù)連接配置，避免硬編碼，示例代碼如下：
# config.py
class Config:
# MySQL數(shù)據(jù)庫(kù)連接信息
MYSQL_HOST = '127.0.0.1' # 數(shù)據(jù)庫(kù)主機(jī)地址（本地為127.0.0.1）
MYSQL_PORT = 3306 # 數(shù)據(jù)庫(kù)端口（默認(rèn)3306）
MYSQL_USER = 'root' # 數(shù)據(jù)庫(kù)用戶名（根據(jù)實(shí)際情況修改）
MYSQL_PASSWORD = '123456' # 數(shù)據(jù)庫(kù)密碼（根據(jù)實(shí)際情況修改）
MYSQL_DB = 'flask_mysql_api'# 數(shù)據(jù)庫(kù)名（與前文創(chuàng)建的一致）
MYSQL_CHARSET = 'utf8mb4' # 字符集
# 開(kāi)發(fā)環(huán)境配置（繼承Config）
class DevelopmentConfig(Config):
DEBUG = True # 開(kāi)啟調(diào)試模式
# 生產(chǎn)環(huán)境配置（繼承Config）
class ProductionConfig(Config):
DEBUG = False # 關(guān)閉調(diào)試模式
# 生產(chǎn)環(huán)境可添加數(shù)據(jù)庫(kù)連接池配置，提升性能
MYSQL_POOL_SIZE = 10
MYSQL_MAX_OVERFLOW = 20
# 配置映射，方便切換環(huán)境
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
- requirements.txt：記錄依賴庫(kù)及版本，執(zhí)行
pip freeze > requirements.txt生成，內(nèi)容示例：
Flask==2.3.3 PyMySQL==1.1.0 Flask-SQLAlchemy==3.1.1
三、基于PyMySQL實(shí)現(xiàn)CRUD API
（一）數(shù)據(jù)庫(kù)連接工具函數(shù)
在app.py中，先編寫數(shù)據(jù)庫(kù)連接和關(guān)閉的工具函數(shù)，復(fù)用連接邏輯，避免重復(fù)代碼：
# app.py
from flask import Flask, request, jsonify
import pymysql
from config import config
# 初始化Flask應(yīng)用
app = Flask(__name__)
# 加載配置（默認(rèn)開(kāi)發(fā)環(huán)境）
app.config.from_object(config['default'])
# 數(shù)據(jù)庫(kù)連接函數(shù)
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 # 游標(biāo)返回字典格式（便于轉(zhuǎn)換為JSON）
)
return connection
except pymysql.Error as e:
app.logger.error(f"數(shù)據(jù)庫(kù)連接失?。簕str(e)}")
return None
# 關(guān)閉數(shù)據(jù)庫(kù)連接函數(shù)
def close_db_connection(connection, cursor):
if cursor:
cursor.close()
if connection:
connection.close()
（二）實(shí)現(xiàn)增刪改查API路由
1. 新增用戶（Create）
- 請(qǐng)求方式：POST
- 請(qǐng)求URL：
/api/users - 請(qǐng)求體：JSON格式，包含
name、age、email字段 - 響應(yīng)：成功返回新增用戶信息（含自增ID），失敗返回錯(cuò)誤信息
@app.route('/api/users', methods=['POST'])
def create_user():
# 獲取請(qǐng)求體JSON數(shù)據(jù)
data = request.get_json()
# 驗(yàn)證必填字段
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': '數(shù)據(jù)庫(kù)連接失敗'}), 500
cursor = connection.cursor()
# 執(zhí)行插入SQL
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
cursor.execute(sql, (data['name'], data['age'], data['email']))
connection.commit() # 提交事務(wù)
# 獲取新增用戶的ID，查詢并返回完整信息
user_id = cursor.lastrowid
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
new_user = cursor.fetchone()
return jsonify({'message': '用戶創(chuàng)建成功', 'user': new_user}), 201 # 201表示創(chuàng)建成功
except pymysql.IntegrityError as e:
# 處理唯一約束沖突（如email重復(fù)）
connection.rollback() # 回滾事務(wù)
return jsonify({'error': '郵箱已存在', 'detail': str(e)}), 409
except Exception as e:
if connection:
connection.rollback()
app.logger.error(f"創(chuàng)建用戶失敗：{str(e)}")
return jsonify({'error': '創(chuàng)建用戶失敗', 'detail': str(e)}), 500
finally:
close_db_connection(connection, cursor)
2. 查詢所有用戶（Read All）
- 請(qǐng)求方式：GET
- 請(qǐng)求URL：
/api/users - 響應(yīng)：返回所有用戶列表（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': '數(shù)據(jù)庫(kù)連接失敗'}), 500
cursor = connection.cursor()
cursor.execute("SELECT * FROM users ORDER BY create_time DESC")
users = cursor.fetchall() # 獲取所有數(shù)據(jù)
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. 查詢單個(gè)用戶（Read One）
- 請(qǐng)求方式：GET
- 請(qǐng)求URL：
/api/users/<int:user_id>（user_id為用戶ID） - 響應(yīng)：成功返回單個(gè)用戶信息，失敗（如用戶不存在）返回錯(cuò)誤信息
@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': '數(shù)據(jù)庫(kù)連接失敗'}), 500
cursor = connection.cursor()
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
user = cursor.fetchone() # 獲取單條數(shù)據(jù)
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）
- 請(qǐng)求方式：PUT
- 請(qǐng)求URL：
/api/users/<int:user_id> - 請(qǐng)求體：JSON格式，包含需更新的字段（如
name、age、email，至少一個(gè)） - 響應(yīng)：成功返回更新后的用戶信息，失敗返回錯(cuò)誤信息
@app.route('/api/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
data = request.get_json()
# 驗(yàn)證至少有一個(gè)可更新字段
update_fields = ['name', 'age', 'email']
if not any(field in data for field in update_fields):
return jsonify({'error': '需提供至少一個(gè)更新字段（name/age/email）'}), 400
connection = None
cursor = None
try:
connection = get_db_connection()
if not connection:
return jsonify({'error': '數(shù)據(jù)庫(kù)連接失敗'}), 500
cursor = connection.cursor()
# 先檢查用戶是否存在
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
if not cursor.fetchone():
return jsonify({'error': '用戶不存在'}), 404
# 構(gòu)建動(dòng)態(tài)更新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）
- 請(qǐng)求方式：DELETE
- 請(qǐng)求URL：
/api/users/<int:user_id> - 響應(yīng)：成功返回刪除確認(rèn)信息，失敗返回錯(cuò)誤信息
@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': '數(shù)據(jù)庫(kù)連接失敗'}), 500
cursor = connection.cursor()
# 檢查用戶是否存在
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
if not cursor.fetchone():
return jsonify({'error': '用戶不存在'}), 404
# 執(zhí)行刪除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)
（三）啟動(dòng)應(yīng)用
在app.py末尾添加啟動(dòng)代碼：
if __name__ == '__main__':
# 開(kāi)發(fā)環(huán)境使用5000端口，生產(chǎn)環(huán)境需修改為80/443或其他端口
app.run(host='0.0.0.0', port=5000, debug=app.config['DEBUG'])
四、基于Flask-SQLAlchemy實(shí)現(xiàn)CRUD API（可選）
Flask-SQLAlchemy是Flask的ORM（對(duì)象關(guān)系映射）擴(kuò)展，可將Python類映射到數(shù)據(jù)庫(kù)表，簡(jiǎn)化SQL操作。以下是簡(jiǎn)化版實(shí)現(xiàn)：
（一）初始化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 # 關(guān)閉SQLAlchemy的修改跟蹤（提升性能）
# 初始化SQLAlchemy
db = SQLAlchemy(app)
# 定義User模型（映射到users表）
class User(db.Model):
__tablename__ = 'users' # 數(shù)據(jù)庫(kù)表名
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())
# 將模型轉(zhuǎn)換為字典（便于返回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')
}
（二）簡(jiǎn)化版CRUD路由示例（以新增和查詢?yōu)槔?/h4>
# 新增用戶（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:
# 創(chuàng)建User對(duì)象（無(wú)需手動(dòng)寫SQL）
new_user = User(
name=data['name'],
age=data['age'],
email=data['email']
)
db.session.add(new_user) # 添加到會(huì)話
db.session.commit() # 提交會(huì)話（等同于SQL的COMMIT）
return jsonify({'message': '用戶創(chuàng)建成功', 'user': new_user.to_dict()}), 201
except Exception as e:
db.session.rollback() # 回滾事務(wù)
return jsonify({'error': '創(chuàng)建用戶失敗', 'detail': str(e)}), 500
# 查詢所有用戶（Flask-SQLAlchemy版）
@app.route('/api/users', methods=['GET'])
def get_all_users():
try:
# 等價(jià)于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
# 其他路由（查詢單個(gè)、更新、刪除）實(shí)現(xiàn)類似，此處省略
# 新增用戶（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:
# 創(chuàng)建User對(duì)象（無(wú)需手動(dòng)寫SQL）
new_user = User(
name=data['name'],
age=data['age'],
email=data['email']
)
db.session.add(new_user) # 添加到會(huì)話
db.session.commit() # 提交會(huì)話（等同于SQL的COMMIT）
return jsonify({'message': '用戶創(chuàng)建成功', 'user': new_user.to_dict()}), 201
except Exception as e:
db.session.rollback() # 回滾事務(wù)
return jsonify({'error': '創(chuàng)建用戶失敗', 'detail': str(e)}), 500
# 查詢所有用戶（Flask-SQLAlchemy版）
@app.route('/api/users', methods=['GET'])
def get_all_users():
try:
# 等價(jià)于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
# 其他路由（查詢單個(gè)、更新、刪除）實(shí)現(xiàn)類似，此處省略
五、API測(cè)試
完成API開(kāi)發(fā)后，需驗(yàn)證各接口功能是否正常，推薦使用Postman工具，測(cè)試步驟如下：
（一）啟動(dòng)應(yīng)用
運(yùn)行app.py：
python app.py
控制臺(tái)顯示* Running on http://0.0.0.0:5000/表示啟動(dòng)成功。
（二）測(cè)試各接口
新增用戶：
- 方法：POST
- URL：
http://127.0.0.1:5000/api/users - 請(qǐng)求體（JSON）：
{ "name": "張三", "age": 25, "email": "zhangsan@example.com" } - 預(yù)期響應(yīng)：狀態(tài)碼201，返回新增用戶信息。
查詢所有用戶：
- 方法：GET
- URL：
http://127.0.0.1:5000/api/users - 預(yù)期響應(yīng)：狀態(tài)碼200，返回包含新增用戶的列表。
查詢單個(gè)用戶：
- 方法：GET
- URL：
http://127.0.0.1:5000/api/users/1（1為新增用戶的ID） - 預(yù)期響應(yīng)：狀態(tài)碼200，返回ID為1的用戶信息。
更新用戶：
- 方法：PUT
- URL：
http://127.0.0.1:5000/api/users/1 - 請(qǐng)求體（JSON）：
{ "age": 26, "email": "zhangsan_update@example.com" } - 預(yù)期響應(yīng)：狀態(tài)碼200，返回更新后的用戶信息（年齡變?yōu)?6，郵箱更新）。
刪除用戶：
- 方法：DELETE
- URL：
http://127.0.0.1:5000/api/users/1 - 預(yù)期響應(yīng)：狀態(tài)碼200，返回刪除成功的提示信息。
六、項(xiàng)目?jī)?yōu)化建議
- 添加請(qǐng)求參數(shù)驗(yàn)證：使用
marshmallow庫(kù)對(duì)請(qǐng)求數(shù)據(jù)進(jìn)行類型、格式驗(yàn)證，提升API健壯性。 - 實(shí)現(xiàn)分頁(yè)查詢：當(dāng)用戶數(shù)據(jù)量大時(shí)，查詢所有用戶可能導(dǎo)致性能問(wèn)題，可添加
page和per_page參數(shù)實(shí)現(xiàn)分頁(yè)。# 分頁(yè)查詢示例（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 # 執(zhí)行帶分頁(yè)的SQL cursor.execute("SELECT * FROM users ORDER BY create_time DESC LIMIT %s OFFSET %s", (per_page, offset)) # ... - 使用數(shù)據(jù)庫(kù)連接池：在生產(chǎn)環(huán)境中，頻繁創(chuàng)建和關(guān)閉數(shù)據(jù)庫(kù)連接會(huì)消耗資源，可使用
DBUtils庫(kù)實(shí)現(xiàn)連接池。 - 添加日志記錄：通過(guò)Flask的
app.logger記錄關(guān)鍵操作和錯(cuò)誤信息，便于問(wèn)題排查。 - 實(shí)現(xiàn)身份認(rèn)證：對(duì)API添加Token驗(yàn)證（如JWT），限制未授權(quán)訪問(wèn)，保護(hù)數(shù)據(jù)安全。
七、完整代碼
下面是使用PyMySQL實(shí)現(xiàn)的完整app.py代碼，可直接運(yùn)行：
from flask import Flask, request, jsonify
import pymysql
# 初始化Flask應(yīng)用
app = Flask(__name__)
# 加載配置（默認(rèn)開(kāi)發(fā)環(huán)境）
# 數(shù)據(jù)庫(kù)連接函數(shù)
def get_db_connection():
try:
connection = pymysql.connect(
host="填寫數(shù)據(jù)庫(kù)ip",
port=端口,
user="用戶",
password="密碼",
db="數(shù)據(jù)庫(kù)"
)
return connection
except pymysql.Error as e:
app.logger.error(f"數(shù)據(jù)庫(kù)連接失?。簕str(e)}")
return None
# 關(guān)閉數(shù)據(jù)庫(kù)連接函數(shù)
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': '數(shù)據(jù)庫(kù)連接失敗'}), 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': '用戶創(chuàng)建成功', '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"創(chuàng)建用戶失?。簕str(e)}")
return jsonify({'error': '創(chuàng)建用戶失敗', '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': '數(shù)據(jù)庫(kù)連接失敗'}), 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)
# 查詢單個(gè)用戶
@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': '數(shù)據(jù)庫(kù)連接失敗'}), 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': '需提供至少一個(gè)更新字段（name/age/email）'}), 400
connection = None
cursor = None
try:
connection = get_db_connection()
if not connection:
return jsonify({'error': '數(shù)據(jù)庫(kù)連接失敗'}), 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': '數(shù)據(jù)庫(kù)連接失敗'}), 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
八、更多細(xì)節(jié)
在Flask框架中，實(shí)現(xiàn)CRUD API的細(xì)節(jié)需要注意以下幾個(gè)方面：
數(shù)據(jù)完整性：在增刪改查操作中，確保數(shù)據(jù)的完整性和一致性。例如，在刪除記錄時(shí)，應(yīng)當(dāng)先記錄原數(shù)據(jù)狀態(tài)，避免數(shù)據(jù)丟失?？梢酝ㄟ^(guò)日志記錄或前后對(duì)比實(shí)現(xiàn)。
事務(wù)管理：在數(shù)據(jù)庫(kù)操作中，使用begin()和commit()確保增刪改查操作屬于同一個(gè)事務(wù)。避免部分執(zhí)行或回滾。在Python中，PyMySQL默認(rèn)支持事務(wù)操作，可以通過(guò)my cur = self.conn.cursor()獲取游標(biāo)，并使用commit()或rollback()控制事務(wù)。
鎖機(jī)制：在高并發(fā)場(chǎng)景下，數(shù)據(jù)庫(kù)鎖機(jī)制可以防止 race condition 問(wèn)題。在Python中，可以通過(guò)self.conn locking實(shí)現(xiàn)鎖獲取，但需要注意鎖的使用場(chǎng)景和時(shí)機(jī)。
錯(cuò)誤處理：在數(shù)據(jù)庫(kù)操作中，必須處理可能出現(xiàn)的錯(cuò)誤，例如ValueError、KeyError、PermissionError等。可以通過(guò)try...except語(yǔ)句捕獲異常，并設(shè)計(jì)合理的錯(cuò)誤處理邏輯。
性能優(yōu)化：在處理大量數(shù)據(jù)時(shí)，需要注意性能優(yōu)化。例如，在查詢中使用索引字段，避免全表掃描；在增刪改查中盡量減少數(shù)據(jù)庫(kù)操作次數(shù)。
九、注意事項(xiàng)
環(huán)境配置：確保Flask框架和PyMySQL庫(kù)的版本與數(shù)據(jù)庫(kù)兼容。避免因版本不兼容導(dǎo)致的功能缺失或異常拋出。
權(quán)限控制：通過(guò)JWT或其他方式實(shí)現(xiàn)用戶認(rèn)證和權(quán)限控制，防止未授權(quán)訪問(wèn)。在API中設(shè)置Token頭，確?？蛻舳朔祷赜行У腡oken。
日志記錄：在關(guān)鍵操作前后添加日志記錄，便于調(diào)試和監(jiān)控?？梢酝ㄟ^(guò)Flask的app.logger實(shí)現(xiàn)日志記錄，確保所有操作都有記錄。
測(cè)試用例：在開(kāi)發(fā)過(guò)程中，編寫詳細(xì)的測(cè)試用例，確保每個(gè)API的功能都能正常工作?？梢酝ㄟ^(guò)Postman、Selenium等工具進(jìn)行自動(dòng)化測(cè)試。
安全性：在API中添加數(shù)據(jù)加密、壓縮機(jī)制，防止敏感數(shù)據(jù)泄露。在傳輸過(guò)程中，使用HTTPS協(xié)議，確保數(shù)據(jù)安全。
十、實(shí)現(xiàn)更多
除了基本的增刪改查，還可以實(shí)現(xiàn)以下更復(fù)雜的功能：
批量操作：在刪除或插入時(shí)，支持批量操作。通過(guò)修改my cur.execute()語(yǔ)句，實(shí)現(xiàn)一次批量處理多個(gè)記錄。
事務(wù)回滾：在數(shù)據(jù)庫(kù)操作中，使用事務(wù)機(jī)制實(shí)現(xiàn)回滾。如果發(fā)生錯(cuò)誤，整個(gè)事務(wù)可以回滾，確保數(shù)據(jù)一致性。
鎖機(jī)制：在高并發(fā)場(chǎng)景下，使用鎖機(jī)制防止數(shù)據(jù)爭(zhēng)用。通過(guò)self.conn.cursor游標(biāo)，將游標(biāo)設(shè)置為鎖模式，實(shí)現(xiàn)鎖機(jī)制。
數(shù)據(jù)校驗(yàn)：在插入或更新時(shí)，對(duì)數(shù)據(jù)進(jìn)行校驗(yàn)。例如，檢查用戶名是否重復(fù)、郵箱格式是否合法等。
十一、拓展思考
其他數(shù)據(jù)庫(kù)：除了MySQL，還可以使用PostgreSQL、MongoDB等其他數(shù)據(jù)庫(kù)。PostgreSQL支持更復(fù)雜的查詢和存儲(chǔ)機(jī)制，適合關(guān)系型和非關(guān)系型數(shù)據(jù)混合存儲(chǔ)場(chǎng)景。
代碼生成工具：使用代碼生成工具（如Flask-RESTful）可以快速生成RESTful API文檔，簡(jiǎn)化API開(kāi)發(fā)流程。
API文檔生成：使用 Swagger 或 Redoc 生成API文檔，確保API的規(guī)范性和可訪問(wèn)性。Flask有內(nèi)置的文檔生成工具，可以自動(dòng)生成 Swagger 格式的文檔。
微服務(wù)架構(gòu)：在開(kāi)發(fā)復(fù)雜應(yīng)用時(shí)，可以采用微服務(wù)架構(gòu)。將不同的功能模塊分離成獨(dú)立的服務(wù)，便于管理和擴(kuò)展。
持續(xù)集成：使用 Jenkins、GitHub Actions 等工具實(shí)現(xiàn)持續(xù)集成，自動(dòng)化測(cè)試和部署過(guò)程。
數(shù)據(jù)庫(kù)遷移：如果需要遷移現(xiàn)有應(yīng)用到新的數(shù)據(jù)庫(kù)或框架，可以使用數(shù)據(jù)庫(kù)遷移工具（如migrate、alembic）自動(dòng)生成遷移腳本，減少人工操作。
十二、總結(jié)
本文詳細(xì)介紹了使用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫(kù)增刪改查API的全過(guò)程，包括環(huán)境搭建、數(shù)據(jù)庫(kù)設(shè)計(jì)、API開(kāi)發(fā)、測(cè)試及優(yōu)化建議。通過(guò)本文的學(xué)習(xí)，你可以掌握：
- Flask框架基本使用方法，包括路由定義、請(qǐng)求處理、響應(yīng)返回。
- PyMySQL庫(kù)連接和操作MySQL數(shù)據(jù)庫(kù)的技巧，如SQL執(zhí)行、事務(wù)處理、異常捕獲。
- RESTful API設(shè)計(jì)規(guī)范，如HTTP方法與CRUD操作的對(duì)應(yīng)關(guān)系、狀態(tài)碼使用。
- 項(xiàng)目?jī)?yōu)化思路，提升API的性能、安全性和可維護(hù)性。
到此這篇關(guān)于Python用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫(kù)的增刪改查API的文章就介紹到這了,更多相關(guān)Flask和PyMySQL實(shí)現(xiàn)MySQL的增刪改查內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
淺談Python3.10 和 Python3.9 之間的差異
多年來(lái)，Python 進(jìn)行了大量升級(jí)，并且在新版本中添加了許多功能。本文就詳細(xì)的介紹 一下Python3.10 和 Python3.9差異，感興趣的朋友可以了解一下2021-09-09
Python使用cProfile分析和定位應(yīng)用性能瓶頸點(diǎn)
cProfile?是?Python?標(biāo)準(zhǔn)庫(kù)中的一個(gè)模塊,用于對(duì)?Python?程序進(jìn)行性能分析,它能輸出每個(gè)函數(shù)的調(diào)用次數(shù)、執(zhí)行耗時(shí)等詳細(xì)信息,下面我們來(lái)看看如何使用cProfile分析和定位應(yīng)用性能瓶頸點(diǎn)吧2024-12-12
pandas改變df列的順序的方法實(shí)現(xiàn)
本文主要介紹了pandas改變df列的順序的方法實(shí)現(xiàn),主要使用 Pandas 中的 reindex() 方法,文中通過(guò)示例代碼介紹的非常詳細(xì),需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2024-03-03
Python3如何解決錯(cuò)誤UnicodeDecodeError
當(dāng)我們使用Python3來(lái)處理文本時(shí),一個(gè)非常常見(jiàn)的問(wèn)題就是UnicodeDecodeError,本文小編就來(lái)深入聊聊這個(gè)錯(cuò)誤是怎么來(lái)的以及怎樣來(lái)解決它吧2025-03-03
Python進(jìn)制轉(zhuǎn)換與反匯編實(shí)現(xiàn)流程介紹
這篇文章主要介紹了Python進(jìn)制轉(zhuǎn)換與反匯編的實(shí)現(xiàn)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧2022-10-10
Linux下通過(guò)python訪問(wèn)MySQL、Oracle、SQL Server數(shù)據(jù)庫(kù)的方法
這篇文章主要介紹了Linux下通過(guò)python訪問(wèn)MySQL、Oracle、SQL Server數(shù)據(jù)庫(kù)的方法,需要的朋友可以參考下2016-04-04
python使用兩種發(fā)郵件的方式smtp和outlook示例
本篇文章主要介紹了python使用兩種發(fā)郵件的方式smtp和outlook示例，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下。2017-06-06
這篇文章主要介紹了django使用html模板減少代碼代碼解析，具有一定借鑒價(jià)值，需要的朋友可以參考下。2017-12-12
今天小編就為大家分享一篇python之消除前綴重命名的方法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-10-10
這篇文章介紹了python通過(guò)ElementTree操作XML的方法，文中通過(guò)示例代碼介紹的非常詳細(xì)。對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-07-07