欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫的增刪改查API

 更新時(shí)間：2025年09月13日 14:18:42   作者：xcLeigh  
Web開發(fā)中,API常需與數(shù)據(jù)庫交互以實(shí)現(xiàn)數(shù)據(jù)的持久化存儲(chǔ),MySQL作為主流關(guān)系型數(shù)據(jù)庫,廣泛用于各類項(xiàng)目,本文基于Flask框架,結(jié)合PyMySQL庫,實(shí)現(xiàn)對MySQL數(shù)據(jù)庫的增刪改查（CRUD）API,適合有基礎(chǔ)Flask知識(shí)和MySQL基礎(chǔ)的開發(fā)者,完整覆蓋環(huán)境搭建、數(shù)據(jù)庫設(shè)計(jì)、API開發(fā)及測試

Python作為一門簡潔、易讀、功能強(qiáng)大的編程語言，其基礎(chǔ)語法是入門學(xué)習(xí)的核心。掌握好基礎(chǔ)語法，能為后續(xù)的編程實(shí)踐打下堅(jiān)實(shí)的基礎(chǔ)。本文將全面講解Python3的基礎(chǔ)語法知識(shí)，適合編程初學(xué)者系統(tǒng)學(xué)習(xí)。Python以其簡潔優(yōu)雅的語法和強(qiáng)大的通用性，成為當(dāng)今最受歡迎的編程語言。本專欄旨在系統(tǒng)性地帶你從零基礎(chǔ)入門到精通Python核心。無論你是零基礎(chǔ)小白還是希望進(jìn)階的專業(yè)開發(fā)者，都將通過清晰的講解、豐富的實(shí)例和實(shí)戰(zhàn)項(xiàng)目，逐步掌握語法基礎(chǔ)、核心數(shù)據(jù)結(jié)構(gòu)、函數(shù)與模塊、面向?qū)ο缶幊?、文件處理、主流庫?yīng)用（如數(shù)據(jù)分析、Web開發(fā)、自動(dòng)化）以及面向?qū)ο蟾呒?jí)特性，最終具備獨(dú)立開發(fā)能力和解決復(fù)雜問題的思維，高效應(yīng)對數(shù)據(jù)分析、人工智能、Web應(yīng)用、自動(dòng)化腳本等廣泛領(lǐng)域的實(shí)際需求。

在Web開發(fā)中，API常需與數(shù)據(jù)庫交互以實(shí)現(xiàn)數(shù)據(jù)的持久化存儲(chǔ)，MySQL作為主流關(guān)系型數(shù)據(jù)庫，廣泛用于各類項(xiàng)目。本文基于Flask框架，結(jié)合PyMySQL庫，實(shí)現(xiàn)對MySQL數(shù)據(jù)庫的增刪改查（CRUD）API，適合有基礎(chǔ)Flask知識(shí)和MySQL基礎(chǔ)的開發(fā)者，完整覆蓋環(huán)境搭建、數(shù)據(jù)庫設(shè)計(jì)、API開發(fā)及測試全流程。

一、項(xiàng)目準(zhǔn)備

（一）技術(shù)棧選擇

  • 后端框架：Flask，輕量級(jí)Web框架，靈活易擴(kuò)展，適合快速開發(fā)API。
  • 數(shù)據(jù)庫：MySQL 8.0及以上版本，穩(wěn)定可靠的關(guān)系型數(shù)據(jù)庫，支持復(fù)雜數(shù)據(jù)查詢與事務(wù)。
  • 數(shù)據(jù)庫連接庫：PyMySQL，純Python實(shí)現(xiàn)的MySQL客戶端庫，用于在Flask項(xiàng)目中連接并操作MySQL數(shù)據(jù)庫。
  • 輔助工具：Postman（API測試）、Navicat/MySQL Workbench（MySQL數(shù)據(jù)庫管理）、VS Code/PyCharm（代碼編寫）。

（二）環(huán)境要求

  1. Python版本：Python 3.6及以上，確保兼容Flask和PyMySQL的最新特性。
  2. 依賴庫安裝：通過pip命令安裝所需庫，命令如下：
# 安裝Flask
pip install flask
# 安裝PyMySQL（連接MySQL）
pip install pymysql
# 安裝Flask-SQLAlchemy（可選，ORM工具，簡化數(shù)據(jù)庫操作）
pip install flask-sqlalchemy

本文先使用PyMySQL原生操作數(shù)據(jù)庫，再補(bǔ)充Flask-SQLAlchemy的實(shí)現(xiàn)方式，滿足不同開發(fā)習(xí)慣需求。

（三）MySQL數(shù)據(jù)庫準(zhǔn)備

  1. 啟動(dòng)MySQL服務(wù)，通過Navicat或MySQL Workbench連接數(shù)據(jù)庫，執(zhí)行以下SQL語句創(chuàng)建項(xiàng)目所需的數(shù)據(jù)庫和表：
-- 創(chuàng)建數(shù)據(jù)庫（命名為flask_mysql_api）
CREATE DATABASE IF NOT EXISTS flask_mysql_api CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

-- 使用數(shù)據(jù)庫
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
);
  1. 驗(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ù)庫連接、邏輯處理）
├── config.py             # 配置文件（數(shù)據(jù)庫連接信息、Flask配置）
└── requirements.txt      # 項(xiàng)目依賴清單（可選，方便他人部署）
  1. config.py：存儲(chǔ)數(shù)據(jù)庫連接配置，避免硬編碼，示例代碼如下：
# config.py
class Config:
    # MySQL數(shù)據(jù)庫連接信息
    MYSQL_HOST = '127.0.0.1'    # 數(shù)據(jù)庫主機(jī)地址（本地為127.0.0.1）
    MYSQL_PORT = 3306           # 數(shù)據(jù)庫端口（默認(rèn)3306）
    MYSQL_USER = 'root'         # 數(shù)據(jù)庫用戶名（根據(jù)實(shí)際情況修改）
    MYSQL_PASSWORD = '123456'   # 數(shù)據(jù)庫密碼（根據(jù)實(shí)際情況修改）
    MYSQL_DB = 'flask_mysql_api'# 數(shù)據(jù)庫名（與前文創(chuàng)建的一致）
    MYSQL_CHARSET = 'utf8mb4'   # 字符集

# 開發(fā)環(huán)境配置（繼承Config）
class DevelopmentConfig(Config):
    DEBUG = True  # 開啟調(diào)試模式

# 生產(chǎn)環(huán)境配置（繼承Config）
class ProductionConfig(Config):
    DEBUG = False  # 關(guān)閉調(diào)試模式
    # 生產(chǎn)環(huán)境可添加數(shù)據(jù)庫連接池配置，提升性能
    MYSQL_POOL_SIZE = 10
    MYSQL_MAX_OVERFLOW = 20

# 配置映射，方便切換環(huán)境
config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}
  1. requirements.txt：記錄依賴庫及版本，執(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ù)庫連接工具函數(shù)

在app.py中，先編寫數(shù)據(jù)庫連接和關(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)開發(fā)環(huán)境）
app.config.from_object(config['default'])

# 數(shù)據(jù)庫連接函數(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ù)庫連接失?。簕str(e)}")
        return None

# 關(guān)閉數(shù)據(jù)庫連接函數(shù)
def close_db_connection(connection, cursor):
    if cursor:
        cursor.close()
    if connection:
        connection.close()

（二）實(shí)現(xiàn)增刪改查API路由

1. 新增用戶（Create）
  • 請求方式：POST
  • 請求URL：/api/users
  • 請求體：JSON格式，包含name、age、email字段
  • 響應(yīng)：成功返回新增用戶信息（含自增ID），失敗返回錯(cuò)誤信息
@app.route('/api/users', methods=['POST'])
def create_user():
    # 獲取請求體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ù)庫連接失敗'}), 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）
  • 請求方式：GET
  • 請求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ù)庫連接失敗'}), 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）
  • 請求方式：GET
  • 請求URL：/api/users/<int:user_id>（user_id為用戶ID）
  • 響應(yīng)：成功返回單個(gè)用戶信息，失?。ㄈ缬脩舨淮嬖冢┓祷劐e(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ù)庫連接失敗'}), 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）
  • 請求方式：PUT
  • 請求URL：/api/users/<int:user_id>
  • 請求體：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ù)庫連接失敗'}), 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）
  • 請求方式：DELETE
  • 請求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ù)庫連接失敗'}), 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__':
    # 開發(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（對象關(guān)系映射）擴(kuò)展，可將Python類映射到數(shù)據(jù)庫表，簡化SQL操作。以下是簡化版實(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ù)庫表名
    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')
        }

（二）簡化版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對象（無需手動(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測試

完成API開發(fā)后，需驗(yàn)證各接口功能是否正常，推薦使用Postman工具，測試步驟如下：

（一）啟動(dòng)應(yīng)用

運(yùn)行app.py：

python app.py

控制臺(tái)顯示* Running on http://0.0.0.0:5000/表示啟動(dòng)成功。

（二）測試各接口

  1. 新增用戶：

    • 方法：POST
    • URL：http://127.0.0.1:5000/api/users
    • 請求體（JSON）：
      {
    "name": "張三",
    "age": 25,
    "email": "zhangsan@example.com"
}
    • 預(yù)期響應(yīng)：狀態(tài)碼201，返回新增用戶信息。

  2. 查詢所有用戶：

    • 方法：GET
    • URL：http://127.0.0.1:5000/api/users
    • 預(yù)期響應(yīng)：狀態(tài)碼200，返回包含新增用戶的列表。

  3. 查詢單個(gè)用戶：

    • 方法：GET
    • URL：http://127.0.0.1:5000/api/users/1（1為新增用戶的ID）
    • 預(yù)期響應(yīng)：狀態(tài)碼200，返回ID為1的用戶信息。

  4. 更新用戶：

    • 方法：PUT
    • URL：http://127.0.0.1:5000/api/users/1
    • 請求體（JSON）：
      {
    "age": 26,
    "email": "zhangsan_update@example.com"
}
    • 預(yù)期響應(yīng)：狀態(tài)碼200，返回更新后的用戶信息（年齡變?yōu)?6，郵箱更新）。

  5. 刪除用戶：

    • 方法：DELETE
    • URL：http://127.0.0.1:5000/api/users/1
    • 預(yù)期響應(yīng)：狀態(tài)碼200，返回刪除成功的提示信息。

六、項(xiàng)目優(yōu)化建議

  1. 添加請求參數(shù)驗(yàn)證：使用marshmallow庫對請求數(shù)據(jù)進(jìn)行類型、格式驗(yàn)證，提升API健壯性。
  2. 實(shí)現(xiàn)分頁查詢：當(dāng)用戶數(shù)據(jù)量大時(shí)，查詢所有用戶可能導(dǎo)致性能問題，可添加page和per_page參數(shù)實(shí)現(xià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í)行帶分頁的SQL
    cursor.execute("SELECT * FROM users ORDER BY create_time DESC LIMIT %s OFFSET %s", (per_page, offset))
    # ...
  3. 使用數(shù)據(jù)庫連接池：在生產(chǎn)環(huán)境中，頻繁創(chuàng)建和關(guān)閉數(shù)據(jù)庫連接會(huì)消耗資源，可使用DBUtils庫實(shí)現(xiàn)連接池。
  4. 添加日志記錄：通過Flask的app.logger記錄關(guān)鍵操作和錯(cuò)誤信息，便于問題排查。
  5. 實(shí)現(xiàn)身份認(rèn)證：對API添加Token驗(yàn)證（如JWT），限制未授權(quá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)開發(fā)環(huán)境）

# 數(shù)據(jù)庫連接函數(shù)
def get_db_connection():
    try:
        connection = pymysql.connect(
            host="填寫數(shù)據(jù)庫ip",
            port=端口,
            user="用戶",
            password="密碼",
            db="數(shù)據(jù)庫"
        )
        return connection
    except pymysql.Error as e:
        app.logger.error(f"數(shù)據(jù)庫連接失敗：{str(e)}")
        return None

# 關(guān)閉數(shù)據(jù)庫連接函數(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ù)庫連接失敗'}), 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ù)庫連接失敗'}), 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ù)庫連接失敗'}), 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ù)庫連接失敗'}), 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ù)庫連接失敗'}), 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ù)丟失?？梢酝ㄟ^日志記錄或前后對比實(shí)現(xiàn)。

事務(wù)管理：在數(shù)據(jù)庫操作中，使用begin()和commit()確保增刪改查操作屬于同一個(gè)事務(wù)。避免部分執(zhí)行或回滾。在Python中，PyMySQL默認(rèn)支持事務(wù)操作，可以通過my cur = self.conn.cursor()獲取游標(biāo)，并使用commit()或rollback()控制事務(wù)。

鎖機(jī)制：在高并發(fā)場景下，數(shù)據(jù)庫鎖機(jī)制可以防止 race condition 問題。在Python中，可以通過self.conn locking實(shí)現(xiàn)鎖獲取，但需要注意鎖的使用場景和時(shí)機(jī)。

錯(cuò)誤處理：在數(shù)據(jù)庫操作中，必須處理可能出現(xiàn)的錯(cuò)誤，例如ValueError、KeyError、PermissionError等。可以通過try...except語句捕獲異常，并設(shè)計(jì)合理的錯(cuò)誤處理邏輯。

性能優(yōu)化：在處理大量數(shù)據(jù)時(shí)，需要注意性能優(yōu)化。例如，在查詢中使用索引字段，避免全表掃描；在增刪改查中盡量減少數(shù)據(jù)庫操作次數(shù)。

九、注意事項(xiàng)

環(huán)境配置：確保Flask框架和PyMySQL庫的版本與數(shù)據(jù)庫兼容。避免因版本不兼容導(dǎo)致的功能缺失或異常拋出。

權(quán)限控制：通過JWT或其他方式實(shí)現(xiàn)用戶認(rèn)證和權(quán)限控制，防止未授權(quán)訪問。在API中設(shè)置Token頭，確?？蛻舳朔祷赜行У腡oken。

日志記錄：在關(guān)鍵操作前后添加日志記錄，便于調(diào)試和監(jiān)控?？梢酝ㄟ^Flask的app.logger實(shí)現(xiàn)日志記錄，確保所有操作都有記錄。

測試用例：在開發(fā)過程中，編寫詳細(xì)的測試用例，確保每個(gè)API的功能都能正常工作?？梢酝ㄟ^Postman、Selenium等工具進(jìn)行自動(dòng)化測試。

安全性：在API中添加數(shù)據(jù)加密、壓縮機(jī)制，防止敏感數(shù)據(jù)泄露。在傳輸過程中，使用HTTPS協(xié)議，確保數(shù)據(jù)安全。

十、實(shí)現(xiàn)更多

除了基本的增刪改查，還可以實(shí)現(xiàn)以下更復(fù)雜的功能：

批量操作：在刪除或插入時(shí)，支持批量操作。通過修改my cur.execute()語句，實(shí)現(xiàn)一次批量處理多個(gè)記錄。

事務(wù)回滾：在數(shù)據(jù)庫操作中，使用事務(wù)機(jī)制實(shí)現(xiàn)回滾。如果發(fā)生錯(cuò)誤，整個(gè)事務(wù)可以回滾，確保數(shù)據(jù)一致性。

鎖機(jī)制：在高并發(fā)場景下，使用鎖機(jī)制防止數(shù)據(jù)爭用。通過self.conn.cursor游標(biāo)，將游標(biāo)設(shè)置為鎖模式，實(shí)現(xiàn)鎖機(jī)制。

數(shù)據(jù)校驗(yàn)：在插入或更新時(shí)，對數(shù)據(jù)進(jìn)行校驗(yàn)。例如，檢查用戶名是否重復(fù)、郵箱格式是否合法等。

十一、拓展思考

其他數(shù)據(jù)庫：除了MySQL，還可以使用PostgreSQL、MongoDB等其他數(shù)據(jù)庫。PostgreSQL支持更復(fù)雜的查詢和存儲(chǔ)機(jī)制，適合關(guān)系型和非關(guān)系型數(shù)據(jù)混合存儲(chǔ)場景。

代碼生成工具：使用代碼生成工具（如Flask-RESTful）可以快速生成RESTful API文檔，簡化API開發(fā)流程。

API文檔生成：使用 Swagger 或 Redoc 生成API文檔，確保API的規(guī)范性和可訪問性。Flask有內(nèi)置的文檔生成工具，可以自動(dòng)生成 Swagger 格式的文檔。

微服務(wù)架構(gòu)：在開發(fā)復(fù)雜應(yīng)用時(shí)，可以采用微服務(wù)架構(gòu)。將不同的功能模塊分離成獨(dú)立的服務(wù)，便于管理和擴(kuò)展。

持續(xù)集成：使用 Jenkins、GitHub Actions 等工具實(shí)現(xiàn)持續(xù)集成，自動(dòng)化測試和部署過程。

數(shù)據(jù)庫遷移：如果需要遷移現(xiàn)有應(yīng)用到新的數(shù)據(jù)庫或框架，可以使用數(shù)據(jù)庫遷移工具（如migrate、alembic）自動(dòng)生成遷移腳本，減少人工操作。

十二、總結(jié)

本文詳細(xì)介紹了使用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫增刪改查API的全過程，包括環(huán)境搭建、數(shù)據(jù)庫設(shè)計(jì)、API開發(fā)、測試及優(yōu)化建議。通過本文的學(xué)習(xí)，你可以掌握：

  1. Flask框架基本使用方法，包括路由定義、請求處理、響應(yīng)返回。
  2. PyMySQL庫連接和操作MySQL數(shù)據(jù)庫的技巧，如SQL執(zhí)行、事務(wù)處理、異常捕獲。
  3. RESTful API設(shè)計(jì)規(guī)范，如HTTP方法與CRUD操作的對應(yīng)關(guān)系、狀態(tài)碼使用。
  4. 項(xiàng)目優(yōu)化思路，提升API的性能、安全性和可維護(hù)性。

 

到此這篇關(guān)于Python用Flask和PyMySQL實(shí)現(xiàn)MySQL數(shù)據(jù)庫的增刪改查API的文章就介紹到這了,更多相關(guān)Flask和PyMySQL實(shí)現(xiàn)MySQL的增刪改查內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

  • 淺談Python3.10 和 Python3.9 之間的差異

    淺談Python3.10 和 Python3.9 之間的差異

    多年來，Python 進(jìn)行了大量升級(jí)，并且在新版本中添加了許多功能。本文就詳細(xì)的介紹 一下Python3.10 和 Python3.9差異，感興趣的朋友可以了解一下
    2021-09-09
  • Python使用cProfile分析和定位應(yīng)用性能瓶頸點(diǎn)

    Python使用cProfile分析和定位應(yīng)用性能瓶頸點(diǎn)

    cProfile?是?Python?標(biāo)準(zhǔn)庫中的一個(gè)模塊,用于對?Python?程序進(jìn)行性能分析,它能輸出每個(gè)函數(shù)的調(diào)用次數(shù)、執(zhí)行耗時(shí)等詳細(xì)信息,下面我們來看看如何使用cProfile分析和定位應(yīng)用性能瓶頸點(diǎn)吧
    2024-12-12
  • pandas改變df列的順序的方法實(shí)現(xiàn)

    pandas改變df列的順序的方法實(shí)現(xiàn)

    本文主要介紹了pandas改變df列的順序的方法實(shí)現(xiàn),主要使用 Pandas 中的 reindex() 方法,文中通過示例代碼介紹的非常詳細(xì),需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2024-03-03
  • Python3如何解決錯(cuò)誤UnicodeDecodeError

    Python3如何解決錯(cuò)誤UnicodeDecodeError

    當(dāng)我們使用Python3來處理文本時(shí),一個(gè)非常常見的問題就是UnicodeDecodeError,本文小編就來深入聊聊這個(gè)錯(cuò)誤是怎么來的以及怎樣來解決它吧
    2025-03-03
  • Python進(jìn)制轉(zhuǎn)換與反匯編實(shí)現(xiàn)流程介紹

    Python進(jìn)制轉(zhuǎn)換與反匯編實(shí)現(xiàn)流程介紹

    這篇文章主要介紹了Python進(jìn)制轉(zhuǎn)換與反匯編的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧
    2022-10-10
  • Linux下通過python訪問MySQL、Oracle、SQL Server數(shù)據(jù)庫的方法

    Linux下通過python訪問MySQL、Oracle、SQL Server數(shù)據(jù)庫的方法

    這篇文章主要介紹了Linux下通過python訪問MySQL、Oracle、SQL Server數(shù)據(jù)庫的方法,需要的朋友可以參考下
    2016-04-04
  • python使用兩種發(fā)郵件的方式smtp和outlook示例

    python使用兩種發(fā)郵件的方式smtp和outlook示例

    本篇文章主要介紹了python使用兩種發(fā)郵件的方式smtp和outlook示例，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下。
    2017-06-06
  • django使用html模板減少代碼代碼解析

    django使用html模板減少代碼代碼解析

    這篇文章主要介紹了django使用html模板減少代碼代碼解析，具有一定借鑒價(jià)值，需要的朋友可以參考下。
    2017-12-12
  • python之消除前綴重命名的方法

    python之消除前綴重命名的方法

    今天小編就為大家分享一篇python之消除前綴重命名的方法，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧
    2018-10-10
  • python通過ElementTree操作XML

    python通過ElementTree操作XML

    這篇文章介紹了python通過ElementTree操作XML的方法，文中通過示例代碼介紹的非常詳細(xì)。對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下
    2022-07-07

最新評論