# 安裝Flask
pip install flask
# 安裝PyMySQL（連接MySQL）
pip install pymysql
# 安裝Flask-SQLAlchemy（可選，ORM工具，簡(jiǎn)化數(shù)據(jù)庫(kù)操作）
pip install flask-sqlalchemy

-- 創(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)建用戶(hù)表（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
);

flask-mysql-api/          # 項(xiàng)目根目錄
├── app.py                # 主程序（API路由、數(shù)據(jù)庫(kù)連接、邏輯處理）
├── config.py             # 配置文件（數(shù)據(jù)庫(kù)連接信息、Flask配置）
└── requirements.txt      # 項(xiàng)目依賴(lài)清單（可選，方便他人部署）

# 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ù)用戶(hù)名（根據(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
}

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

# 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()

@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ù)

        # 獲取新增用戶(hù)的ID，查詢(xún)并返回完整信息
        user_id = cursor.lastrowid
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        new_user = cursor.fetchone()

        return jsonify({'message': '用戶(hù)創(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)建用戶(hù)失?。簕str(e)}")
        return jsonify({'error': '創(chuàng)建用戶(hù)失敗', '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()  # 獲取所有數(shù)據(jù)

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

    except Exception as e:
        app.logger.error(f"查詢(xún)所有用戶(hù)失?。簕str(e)}")
        return jsonify({'error': '查詢(xún)用戶(hù)失敗', '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': '數(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': '用戶(hù)不存在'}), 404

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

    except Exception as e:
        app.logger.error(f"查詢(xún)用戶(hù){user_id}失敗：{str(e)}")
        return jsonify({'error': '查詢(xún)用戶(hù)失敗', '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()
    # 驗(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()
        # 先檢查用戶(hù)是否存在
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用戶(hù)不存在'}), 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()

        # 查詢(xún)更新后的用戶(hù)信息
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        updated_user = cursor.fetchone()

        return jsonify({'message': '用戶(hù)更新成功', '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"更新用戶(hù){user_id}失?。簕str(e)}")
        return jsonify({'error': '更新用戶(hù)失敗', '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()
        # 檢查用戶(hù)是否存在
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        if not cursor.fetchone():
            return jsonify({'error': '用戶(hù)不存在'}), 404

        # 執(zhí)行刪除SQL
        cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
        connection.commit()

        return jsonify({'message': f'用戶(hù){user_id}刪除成功'}), 200

    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"刪除用戶(hù){user_id}失?。簕str(e)}")
        return jsonify({'error': '刪除用戶(hù)失敗', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

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'])

# 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')
        }

python app.py

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

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

# 分頁(yè)查詢(xún)示例（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))
    # ...

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="填寫(xiě)數(shù)據(jù)庫(kù)ip",
            port=端口,
            user="用戶(hù)",
            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()

# 新增用戶(hù)
@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': '用戶(hù)創(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)建用戶(hù)失敗：{str(e)}")
        return jsonify({'error': '創(chuàng)建用戶(hù)失敗', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 查詢(xún)所有用戶(hù)
@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"查詢(xún)所有用戶(hù)失敗：{str(e)}")
        return jsonify({'error': '查詢(xún)用戶(hù)失敗', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 查詢(xún)單個(gè)用戶(hù)
@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': '用戶(hù)不存在'}), 404

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

    except Exception as e:
        app.logger.error(f"查詢(xún)用戶(hù){user_id}失?。簕str(e)}")
        return jsonify({'error': '查詢(xún)用戶(hù)失敗', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 更新用戶(hù)
@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': '用戶(hù)不存在'}), 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': '用戶(hù)更新成功', '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"更新用戶(hù){user_id}失敗：{str(e)}")
        return jsonify({'error': '更新用戶(hù)失敗', 'detail': str(e)}), 500
    finally:
        close_db_connection(connection, cursor)

# 刪除用戶(hù)
@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': '用戶(hù)不存在'}), 404

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

        return jsonify({'message': f'用戶(hù){user_id}刪除成功'}), 200

    except Exception as e:
        if connection:
            connection.rollback()
        app.logger.error(f"刪除用戶(hù){user_id}失敗：{str(e)}")
        return jsonify({'error': '刪除用戶(hù)失敗', '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'])