最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

利用Flask實現(xiàn)MySQL數(shù)據(jù)庫增刪改查API的全過程

 更新時間：2025年09月09日 09:03:38   作者：xcLeigh  
這篇文章主要介紹了如何用Flask實現(xiàn)MySQL數(shù)據(jù)庫增刪改查API,本文圍繞用Flask實現(xiàn)MySQL數(shù)據(jù)庫增刪改查（CRUD）API展開,并附上完整代碼示例,幫助開發(fā)者掌握Flask結(jié)合MySQL開發(fā)API的方法,需要的朋友可以參考下

前言

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

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

一、項目準備

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

  • 后端框架：Flask，輕量級Web框架，靈活易擴展，適合快速開發(fā)API。
  • 數(shù)據(jù)庫：MySQL 8.0及以上版本，穩(wěn)定可靠的關(guān)系型數(shù)據(jù)庫，支持復雜數(shù)據(jù)查詢與事務。
  • 數(shù)據(jù)庫連接庫：PyMySQL，純Python實現(xiàn)的MySQL客戶端庫，用于在Flask項目中連接并操作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ù)庫，再補充Flask-SQLAlchemy的實現(xiàn)方式，滿足不同開發(fā)習慣需求。

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

  1. 啟動MySQL服務，通過Navicat或MySQL Workbench連接數(shù)據(jù)庫，執(zhí)行以下SQL語句創(chuà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. 驗證表結(jié)構(gòu)：執(zhí)行DESCRIBE users;，確認字段名、類型、約束是否正確。

二、項目結(jié)構(gòu)搭建

為保證代碼可讀性和可維護性，創(chuàng)建如下項目目錄結(jié)構(gòu)：

flask-mysql-api/          # 項目根目錄
├── app.py                # 主程序（API路由、數(shù)據(jù)庫連接、邏輯處理）
├── config.py             # 配置文件（數(shù)據(jù)庫連接信息、Flask配置）
└── requirements.txt      # 項目依賴清單（可選，方便他人部署）
  1. config.py：存儲數(shù)據(jù)庫連接配置，避免硬編碼，示例代碼如下：
# config.py
class Config:
    # MySQL數(shù)據(jù)庫連接信息
    MYSQL_HOST = '127.0.0.1'    # 數(shù)據(jù)庫主機地址（本地為127.0.0.1）
    MYSQL_PORT = 3306           # 數(shù)據(jù)庫端口（默認3306）
    MYSQL_USER = 'root'         # 數(shù)據(jù)庫用戶名（根據(jù)實際情況修改）
    MYSQL_PASSWORD = '123456'   # 數(shù)據(jù)庫密碼（根據(jù)實際情況修改）
    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實現(xiàn)CRUD API

（一）數(shù)據(jù)庫連接工具函數(shù)

在app.py中，先編寫數(shù)據(jù)庫連接和關(guān)閉的工具函數(shù)，復用連接邏輯，避免重復代碼：

# app.py
from flask import Flask, request, jsonify
import pymysql
from config import config

# 初始化Flask應用
app = Flask(__name__)
# 加載配置（默認開發(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  # 游標返回字典格式（便于轉(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()

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

1. 新增用戶（Create）

  • 請求方式：POST
  • 請求URL：/api/users
  • 請求體：JSON格式，包含name、age、email字段
  • 響應：成功返回新增用戶信息（含自增ID），失敗返回錯誤信息
@app.route('/api/users', methods=['POST'])
def create_user():
    # 獲取請求體JSON數(shù)據(jù)
    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()
        # 執(zhí)行插入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': '用戶創(chuàng)建成功', 'user': new_user}), 201  # 201表示創(chuàng)建成功

    except pymysql.IntegrityError as e:
        # 處理唯一約束沖突（如email重復）
        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)

2. 查詢所有用戶（Read All）

  • 請求方式：GET
  • 請求URL：/api/users
  • 響應：返回所有用戶列表（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. 查詢單個用戶（Read One）

  • 請求方式：GET
  • 請求URL：/api/users/<int:user_id>（user_id為用戶ID）
  • 響應：成功返回單個用戶信息，失?。ㄈ缬脩舨淮嬖冢┓祷劐e誤信息
@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，至少一個）
  • 響應：成功返回更新后的用戶信息，失敗返回錯誤信息
@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': '需提供至少一個更新字段（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)建動態(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>
  • 響應：成功返回刪除確認信息，失敗返回錯誤信息
@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)

（三）啟動應用

在app.py末尾添加啟動代碼：

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實現(xiàn)CRUD API（可選）

Flask-SQLAlchemy是Flask的ORM（對象關(guān)系映射）擴展，可將Python類映射到數(shù)據(jù)庫表，簡化SQL操作。以下是簡化版實現(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)槔?/h3> 
# 新增用戶（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對象（無需手動寫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': '用戶創(chuàng)建成功', 'user': new_user.to_dict()}), 201
    except Exception as e:
       db.session.rollback()  # 回滾事務
        return jsonify({'error': '創(chuàng)建用戶失敗', 'detail': str(e)}), 500

# 查詢所有用戶（Flask-SQLAlchemy版）
@app.route('/api/users', methods=['GET'])
def get_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]
        }), 200
    except Exception as e:
        return jsonify({'error': '查詢用戶失敗', 'detail': str(e)}), 500

# 其他路由（查詢單個、更新、刪除）實現(xiàn)類似，此處省略

五、API測試

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

（一）啟動應用

運行app.py：

python app.py

控制臺顯示* Running on http://0.0.0.0:5000/表示啟動成功。

（二）測試各接口

新增用戶：

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

查詢所有用戶：

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

查詢單個用戶：

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

更新用戶：

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

刪除用戶：

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

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

  1. 添加請求參數(shù)驗證：使用marshmallow庫對請求數(shù)據(jù)進行類型、格式驗證，提升API健壯性。
  2. 實現(xiàn)分頁查詢：當用戶數(shù)據(jù)量大時，查詢所有用戶可能導致性能問題，可添加page和per_page參數(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))
    # ...
  1. 使用數(shù)據(jù)庫連接池：在生產(chǎn)環(huán)境中，頻繁創(chuàng)建和關(guān)閉數(shù)據(jù)庫連接會消耗資源，可使用DBUtils庫實現(xiàn)連接池。
  2. 添加日志記錄：通過Flask的app.logger記錄關(guān)鍵操作和錯誤信息，便于問題排查。
  3. 實現(xiàn)身份認證：對API添加Token驗證（如JWT），限制未授權(quán)訪問，保護數(shù)據(jù)安全。

七、完整代碼

下面是使用PyMySQL實現(xiàn)的完整app.py代碼，可直接運行：

from flask import Flask, request, jsonify
import pymysql

# 初始化Flask應用
app = Flask(__name__)
# 加載配置（默認開發(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)

# 查詢單個用戶
@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': '需提供至少一個更新字段（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

八、總結(jié)

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

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

后續(xù)可進一步擴展功能，如添加用戶認證、實現(xiàn)更復雜的查詢邏輯、集成前端頁面等，逐步完善項目。

以上就是利用Flask實現(xiàn)MySQL數(shù)據(jù)庫增刪改查API的完整代碼的詳細內(nèi)容，更多關(guān)于Flask實現(xiàn)MySQL增刪改查API的資料請關(guān)注腳本之家其它相關(guān)文章！

相關(guān)文章

  • python實現(xiàn)順序表的簡單代碼

    python實現(xiàn)順序表的簡單代碼

    這篇文章主要為大家詳細介紹了順序表定義及python實現(xiàn)代碼，具有一定的參考價值，感興趣的小伙伴們可以參考一下
    2018-09-09
  • 解決python中0x80072ee2錯誤的方法

    解決python中0x80072ee2錯誤的方法

    在本篇文章中小編給大家分享的是關(guān)于解決python中0x80072ee2錯誤的方法，需要的朋友們可以參考下。
    2020-07-07
  • Python面向?qū)ο髮崿F(xiàn)一個對象調(diào)用另一個對象操作示例

    Python面向?qū)ο髮崿F(xiàn)一個對象調(diào)用另一個對象操作示例

    這篇文章主要介紹了Python面向?qū)ο髮崿F(xiàn)一個對象調(diào)用另一個對象操作,結(jié)合實例形式分析了Python對象的定義、初始化、調(diào)用等相關(guān)操作技巧,需要的朋友可以參考下
    2019-04-04
  • python實現(xiàn)簡單圖片物體標注工具

    python實現(xiàn)簡單圖片物體標注工具

    這篇文章主要為大家詳細介紹了python實現(xiàn)簡單圖片物體標注工具，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
    2019-03-03
  • Django 返回json數(shù)據(jù)的實現(xiàn)示例

    Django 返回json數(shù)據(jù)的實現(xiàn)示例

    這篇文章主要介紹了Django 返回json數(shù)據(jù)的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
    2020-03-03
  • PyGame實現(xiàn)初始化導入所有模塊方法詳解

    PyGame實現(xiàn)初始化導入所有模塊方法詳解

    pygame安裝是為了開發(fā)小游戲,在下新手在經(jīng)過許多嘗試后,為大家避雷,給大家分享一個簡單有效的方法,下面這篇文章主要給大家介紹了關(guān)于Python中Pygame的詳細安裝過程的相關(guān)資料,需要的朋友可以參考下
    2022-11-11
  • Python繪制正二十面體圖形示例

    Python繪制正二十面體圖形示例

    正二十面體由20個小的正三角形面組成,每個頂點周圍有?5?個頂點,下面這篇文章主要給大家介紹了關(guān)于Python繪制正二十面體圖形的相關(guān)資料,需要的朋友可以參考下
    2022-12-12
  • 實例講解Python中SocketServer模塊處理網(wǎng)絡請求的用法

    實例講解Python中SocketServer模塊處理網(wǎng)絡請求的用法

    SocketServer模塊中帶有很多實現(xiàn)服務器所能夠用到的socket類和操作方法,下面我們就來以實例講解Python中SocketServer模塊處理網(wǎng)絡請求的用法:
    2016-06-06
  • OpenCV半小時掌握基本操作之角點檢測

    OpenCV半小時掌握基本操作之角點檢測

    這篇文章主要介紹了OpenCV基本操作之角點檢測，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
    2021-09-09
  • Python 圖片文字識別的實現(xiàn)之PaddleOCR

    Python 圖片文字識別的實現(xiàn)之PaddleOCR

    OCR方向的工程師，之前一定聽說過PaddleOCR這個項目，其主要推薦的PP-OCR算法更是被國內(nèi)外企業(yè)開發(fā)者廣泛應用，短短半年時間，累計Star數(shù)量已超過15k，頻頻登上Github Trending和Paperswithcode 日榜月榜第一
    2021-11-11

最新評論

伊春市| 年辖：市辖区| 墨脱县| 平安县| 都昌县| 罗定市| 宕昌县| 台东市| 阿克陶县| 海兴县| 饶阳县| 布拖县| 温州市| 通山县| 铅山县| 广饶县| 宿迁市| 满城县| 光山县| 江北区| 鄂尔多斯市| 怀来县| 竹北市| 始兴县| 福建省| 屯留县| 云龙县| 昌邑市| 布尔津县| 临澧县| 封丘县| 岳阳市| 银川市| 武陟县| 平昌县| 黄山市| 神木县| 繁昌县| 抚宁县| 阜阳市| 安溪县|