Python Web开发：用Flask实现MySQL数据库的增删改查接口

本文将详细介绍如何使用轻量级Web框架Flask构建完整的MySQL数据库操作接口。通过清晰的步骤和完整代码示例，您将掌握从环境配置到API测试的全流程开发技能。

---

1. 环境准备

安装必要依赖：

pip install flask flask-mysqldb

创建项目结构：

/myapp
  ├── app.py         # 主程序
  ├── config.py      # 配置参数
  ├── requirements.txt

---

2. 数据库配置

config.py 文件配置MySQL连接：

class Config:
    MYSQL_HOST = 'localhost'
    MYSQL_USER = 'root'
    MYSQL_PASSWORD = 'your_password'
    MYSQL_DB = 'flask_demo'
    MYSQL_CURSORCLASS = 'DictCursor'  # 返回字典格式数据

---

3. 初始化Flask应用

app.py 基础框架：

from flask import Flask, jsonify, request
from flask_mysqldb import MySQL
import config

app = Flask(__name__)
app.config.from_object(config.Config)
mysql = MySQL(app)

# 创建users表
@app.before_first_request
def init_db():
    cur = mysql.connection.cursor()
    cur.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(50) NOT NULL,
            email VARCHAR(100) UNIQUE NOT NULL
        )
    ''')
    mysql.connection.commit()

---

4. 实现CRUD接口

4.1 创建数据 (Create)

@app.route('/users', methods=['POST'])
def add_user():
    data = request.get_json()
    cur = mysql.connection.cursor()
    cur.execute("INSERT INTO users(name, email) VALUES (%s, %s)", 
                (data['name'], data['email']))
    mysql.connection.commit()
    return jsonify({"message": "User created"}), 201

4.2 读取数据 (Read)

@app.route('/users', methods=['GET'])
def get_users():
    cur = mysql.connection.cursor()
    cur.execute("SELECT * FROM users")
    users = cur.fetchall()
    return jsonify(users)

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    cur = mysql.connection.cursor()
    cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    user = cur.fetchone()
    return jsonify(user) if user else ('', 404)

4.3 更新数据 (Update)

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    cur = mysql.connection.cursor()
    cur.execute("UPDATE users SET name=%s, email=%s WHERE id=%s", 
                (data['name'], data['email'], user_id))
    mysql.connection.commit()
    return jsonify({"message": "User updated"})

4.4 删除数据 (Delete)

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    cur = mysql.connection.cursor()
    cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
    mysql.connection.commit()
    return jsonify({"message": "User deleted"})

---

5. 接口测试

使用curl测试API：

# 创建用户
curl -X POST http://localhost:5000/users \
     -H "Content-Type: application/json" \
     -d '{"name":"Alice", "email":"alice@example.com"}'

# 查询所有用户
curl http://localhost:5000/users

# 更新用户
curl -X PUT http://localhost:5000/users/1 \
     -d '{"name":"Alice Smith", "email":"alice.smith@example.com"}'

# 删除用户
curl -X DELETE http://localhost:5000/users/1

---

6. 关键优化建议

1. 错误处理：添加数据库操作异常捕获

   try:
       # 数据库操作
   except Exception as e:
       return jsonify({"error": str(e)}), 500
   

2. 参数校验：使用Flask-Inputs验证请求数据

   from flask_inputs import Inputs
   from wtforms.validators import DataRequired, Email
   
   class UserInputs(Inputs):
       json = {
           'name': [DataRequired()],
           'email': [DataRequired(), Email()]
       }
   

3. 连接池：配置MYSQL_POOL_SIZE提升并发性能

---

7. 总结

本文实现了基于Flask的完整MySQL CRUD接口，包含：

• 使用flask-mysqldb集成MySQL
• RESTful风格API设计
• 请求参数处理与响应标准化
• 基础错误处理机制

通过扩展用户认证、分页查询等功能，可快速构建生产级Web服务。完整代码示例可在GitHub仓库获取。

技术栈扩展建议：

• 使用SQLAlchemy替代原生SQL
• 集成Swagger实现API文档自动化
• 添加JWT用户认证模块