程式解惑室：羅煜鈞的開發日誌與技術分享: FastAPI 電商訂單系統：高效能 RESTful API 實戰

根據 database_setup.sql 資料庫結構，以下是一個基於 FastAPI的 RESTful API 實現。此代碼旨在滿足電商訂單 API 功能要求、部分效能考量和錯誤處理，並對照您提供的初始代碼進行了修正和增強。

專案設定 (概念)

安裝 FastAPI 和 Uvicorn:

Bash

pip install fastapi uvicorn "python-dotenv" "mysql-connector-python"

資料庫: 我們將根據 database_setup.sql 檔案假定使用 MySQL 資料庫。我們將使用 mysql-connector-python 進行資料庫互動。
環境變數: 對於敏感資訊（如資料庫憑證），我們將使用 .env 檔案和 python-dotenv。

FastAPI 實作

這是 FastAPI 應用程式的 Python 程式碼。

1. main.py (FastAPI 應用程式)

Python
# main.py
import os
from datetime import datetime
from math import ceil
from typing import List, Dict, Union, Any

from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import Error

# 從 .env 檔案載入環境變數
load_dotenv()

app = FastAPI(
    title="電子商務訂單管理 API",
    description="用於管理訂單、產品和訂單統計的 API。",
    version="1.0.0"
)

# 資料庫連線設定
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "ecommerce_test")
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "") # 在實際生產環境中，請使用環境變數

def get_db_connection():
    """建立並回傳資料庫連線。"""
    conn = None
    try:
        conn = mysql.connector.connect(
            host=DB_HOST,
            database=DB_NAME,
            user=DB_USER,
            password=DB_PASSWORD
        )
        if conn.is_connected():
            print("成功連接到資料庫")
        return conn
    except Error as e:
        print(f"連接 MySQL 資料庫時發生錯誤: {e}")
        # 在生產環境中，更穩健地記錄此錯誤
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="伺服器錯誤: 無法連接資料庫"
        )

# Pydantic 模型用於請求和回應驗證/序列化
class OrderItemRequest(BaseModel):
    product_id: int = Field(..., gt=0, description="產品 ID")
    quantity: int = Field(..., gt=0, description="產品數量")

class CreateOrderRequest(BaseModel):
    user_id: int = Field(..., gt=0, description="下訂單的使用者 ID")
    items: List[OrderItemRequest] = Field(..., min_items=1, description="訂單中的商品清單")

class OrderResponse(BaseModel):
    id: int
    user_id: int
    number: str
    status: str
    total_amount: float
    created_at: datetime

class OrderListResponse(BaseModel):
    data: List[OrderResponse]
    page: int
    total_pages: int
    total_items: int

class ProductResponse(BaseModel):
    id: int
    name: str
    price: float
    stock: int

class ProductListResponse(BaseModel):
    data: List[ProductResponse]
    page: int
    total_pages: int
    total_items: int

class OrderStatsResponse(BaseModel):
    total_orders: int
    total_amount: float
    today_orders: int
    today_amount: float

class UpdateOrderStatusRequest(BaseModel):
    status: str = Field(..., pattern="^(pending|processing|shipped|delivered|cancelled)$", description="訂單的新狀態")

# 依賴項，用於獲取資料庫連線並確保其關閉
def get_db():
    db = get_db_connection()
    try:
        yield db
    finally:
        if db:
            db.close()

# API 路由

@app.get("/api/orders", response_model=OrderListResponse, summary="獲取分頁的訂單列表")
async def get_orders(page: int = 1, limit: int = 20, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    擷取分頁的訂單列表。
    """
    if limit <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="限制必須是正整數。")
    if page <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="頁碼必須是正整數。")

    offset = (page - 1) * limit
    cursor = db.cursor(dictionary=True)

    try:
        # 獲取總數
        cursor.execute("SELECT COUNT(*) FROM orders")
        total_items = cursor.fetchone()['COUNT(*)']
        total_pages = ceil(total_items / limit) if total_items > 0 else 0

        # 獲取分頁訂單
        cursor.execute(
            "SELECT id, user_id, number, status, total_amount, created_at FROM orders LIMIT %s OFFSET %s",
            (limit, offset)
        )
        orders = cursor.fetchall()

        return {
            "data": orders,
            "page": page,
            "total_pages": total_pages,
            "total_items": total_items
        }
    except Error as e:
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法獲取訂單列表")
    finally:
        cursor.close()


@app.get("/api/orders/{order_id}", response_model=OrderResponse, summary="獲取單一訂單的詳細資訊")
async def get_order_details(order_id: int, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    根據訂單 ID 擷取特定訂單的詳細資訊。
    """
    cursor = db.cursor(dictionary=True)
    try:
        cursor.execute("SELECT id, user_id, number, status, total_amount, created_at FROM orders WHERE id = %s", (order_id,))
        order = cursor.fetchone()
        if not order:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="訂單不存在")
        return order
    except Error as e:
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法獲取訂單詳情")
    finally:
        cursor.close()


@app.post("/api/orders", status_code=status.HTTP_201_CREATED, summary="建立新訂單")
async def create_order(order_data: CreateOrderRequest, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    建立新訂單，扣除產品庫存並管理交易。
    """
    cursor = db.cursor()
    try:
        db.start_transaction()

        user_id = order_data.user_id
        # 生成唯一的訂單號 (類似 PHP 的邏輯)
        order_number = f"ORD{datetime.now().strftime('%Y%m%d%H%M%S')}{os.urandom(2).hex()}" # 比 mt_rand 更健壯
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

        # 插入 orders 表
        cursor.execute(
            "INSERT INTO orders (user_id, number, status, created_at, updated_at) VALUES (%s, %s, 'pending', %s, %s)",
            (user_id, order_number, current_time, current_time)
        )
        order_id = cursor.lastrowid
        if not order_id:
            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="無法創建訂單")

        total_amount = 0.0

        for item in order_data.items:
            product_id = item.product_id
            quantity = item.quantity

            # 鎖定產品行以進行更新，防止競爭條件 (FOR UPDATE 等效)
            # 在 mysql.connector 中，通常透過確保事務隱式持有鎖定來處理此問題，透過 UPDATE/SELECT FOR UPDATE。
            # 這裡我們將先進行 SELECT，然後進行 UPDATE，依賴於事務。
            cursor.execute("SELECT stock, price, is_deleted FROM products WHERE id = %s FOR UPDATE", (product_id,))
            product = cursor.fetchone()

            if not product:
                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"產品 ID {product_id} 不存在")
            
            product_stock, product_price, is_deleted = product[0], product[1], product[2] # 透過索引存取非字典游標

            if is_deleted:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"產品 ID {product_id} 已停用或刪除")

            if product_stock < quantity:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"產品 ID {product_id} 庫存不足。可用: {product_stock}, 需求: {quantity}")

            unit_price = float(product_price)
            subtotal = unit_price * quantity
            total_amount += subtotal

            # 更新產品庫存
            cursor.execute(
                "UPDATE products SET stock = stock - %s, updated_at = %s WHERE id = %s AND stock >= %s",
                (quantity, current_time, product_id, quantity)
            )
            if cursor.rowcount == 0:
                 # 如果初始檢查後庫存不足，則可能會發生這種情況，因為存在競爭條件
                db.rollback() # 在拋出異常前明確回滾
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"產品 ID {product_id} 庫存更新失敗，可能庫存不足或並發問題")

            # 插入 order_items
            cursor.execute(
                "INSERT INTO order_items (order_id, product_id, quantity, unit_price, subtotal, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s)",
                (order_id, product_id, quantity, unit_price, subtotal, current_time, current_time)
            )

        # 更新 orders 表中的總金額
        cursor.execute("UPDATE orders SET total_amount = %s WHERE id = %s", (total_amount, order_id))

        db.commit()
        return {"order_id": order_id, "order_number": order_number}

    except HTTPException as e:
        db.rollback()
        raise e # 重新拋出 FastAPI HTTPException
    except Error as e:
        db.rollback()
        print(f"訂單創建期間的資料庫錯誤: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"伺服器錯誤: 無法創建訂單。詳細錯誤: {e}" # 開發環境中更詳細，生產環境中更通用
        )
    except Exception as e:
        db.rollback()
        print(f"訂單創建期間發生意外錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"伺服器錯誤: {e}")
    finally:
        cursor.close()


@app.put("/api/orders/{order_id}/status", summary="更新訂單狀態")
async def update_order_status(order_id: int, status_data: UpdateOrderStatusRequest, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    更新特定訂單的狀態。
    允許的狀態: 'pending'、'processing'、'shipped'、'delivered'、'cancelled'。
    """
    cursor = db.cursor()
    try:
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        cursor.execute(
            "UPDATE orders SET status = %s, updated_at = %s WHERE id = %s",
            (status_data.status, current_time, order_id)
        )
        if cursor.rowcount == 0:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="訂單不存在或狀態已是目標狀態")
        db.commit()
        return {"success": True, "message": "訂單狀態更新成功"}
    except Error as e:
        db.rollback()
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法更新訂單狀態")
    finally:
        cursor.close()


@app.get("/api/orders/stats", response_model=OrderStatsResponse, summary="獲取訂單統計")
async def get_order_stats(db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    擷取訂單的聚合統計資訊，包括總訂單數、總金額、今日訂單數和今日金額。
    """
    cursor = db.cursor(dictionary=True)
    try:
        cursor.execute("""
            SELECT
                (SELECT COUNT(*) FROM orders) as total_orders,
                (SELECT COALESCE(SUM(total_amount), 0) FROM orders) as total_amount,
                (SELECT COUNT(*) FROM orders WHERE DATE(created_at) = CURDATE()) as today_orders,
                (SELECT COALESCE(SUM(total_amount), 0) FROM orders WHERE DATE(created_at) = CURDATE()) as today_amount
        """)
        stats = cursor.fetchone()

        # 確保正確的類型並處理如果沒有訂單時總和可能為 None 的情況
        return OrderStatsResponse(
            total_orders=int(stats['total_orders']),
            total_amount=float(stats['total_amount']),
            today_orders=int(stats['today_orders']),
            today_amount=float(stats['today_amount'])
        )
    except Error as e:
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法獲取訂單統計")
    finally:
        cursor.close()


@app.get("/api/products", response_model=ProductListResponse, summary="獲取分頁的產品列表")
async def get_products(page: int = 1, limit: int = 50, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    """
    擷取活躍產品 (is_deleted = FALSE) 的分頁列表。
    """
    if limit <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="限制必須是正整數。")
    if page <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="頁碼必須是正整數。")

    offset = (page - 1) * limit
    cursor = db.cursor(dictionary=True)

    try:
        # 獲取活躍產品的總數
        cursor.execute("SELECT COUNT(*) FROM products WHERE is_deleted = FALSE")
        total_items = cursor.fetchone()['COUNT(*)']
        total_pages = ceil(total_items / limit) if total_items > 0 else 0

        # 獲取分頁產品
        cursor.execute(
            "SELECT id, name, price, stock FROM products WHERE is_deleted = FALSE LIMIT %s OFFSET %s",
            (limit, offset)
        )
        products = cursor.fetchall()

        return {
            "data": products,
            "page": page,
            "total_pages": total_pages,
            "total_items": total_items
        }
    except Error as e:
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法獲取產品列表")
    finally:
        cursor.close()

2. .env 檔案 (用於資料庫憑證)

在 main.py 相同的目錄中建立一個名為 .env 的檔案:

DB_HOST=localhost
DB_NAME=ecommerce_test
DB_USER=root
DB_PASSWORD=

3. database_setup.sql (已在提示中提供，執行此檔案以設定您的資料庫)

此檔案已在您的提示中提供。您應該在您的 MySQL 伺服器上執行此 SQL 腳本，以建立 ecommerce_test 資料庫和必要的表格 (users、products、orders、order_items)。

如何執行 FastAPI 應用程式

儲存檔案: 將 Python 程式碼儲存為 main.py，並將環境變數儲存為 .env 在同一目錄中。確保您的 database_setup.sql 已準備就緒。
設定資料庫: 確保您的 MySQL 正在執行，並且已執行 database_setup.sql 以建立 ecommerce_test 資料庫。

安裝依賴項:

Bash

pip install fastapi uvicorn "python-dotenv" "mysql-connector-python"

執行應用程式:
Bash
uvicorn main:app --reload
這將啟動 Uvicorn 伺服器，通常在 http://127.0.0.1:8000 上。--reload 標誌將在程式碼更改時自動重啟伺服器。

測試 API

您可以透過 http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc 存取互動式 API 文件。

使用 curl 的範例:

建立訂單 (POST /api/orders):

Bash
curl -X POST "http://127.0.0.1:8000/api/orders" \
     -H "Content-Type: application/json" \
     -d '{
           "user_id": 1,
           "items": [
             {"product_id": 1, "quantity": 2},
             {"product_id": 3, "quantity": 1}
           ]
         }'

(注意: 您需要在資料庫中擁有 user_id 1 以及 product_id 1 和 3，並有足夠的庫存才能使其正常運作。)

獲取訂單 (GET /api/orders):

Bash

curl "http://127.0.0.1:8000/api/orders?page=1&limit=20"

獲取訂單詳細資訊 (GET /api/orders/{order_id}):

Bash

curl "http://127.0.0.1:8000/api/orders/1" # 將 1 替換為實際的訂單 ID

更新訂單狀態 (PUT /api/orders/{order_id}/status):

Bash
curl -X PUT "http://127.0.0.1:8000/api/orders/1/status" \
     -H "Content-Type: application/json" \
     -d '{"status": "shipped"}'

獲取訂單統計 (GET /api/orders/stats):

Bash

curl "http://127.0.0.1:8000/api/orders/stats"

獲取產品 (GET /api/products):

Bash

curl "http://127.0.0.1:8000/api/products?page=1&limit=50"

使用的主要 FastAPI 功能:

Pydantic 模型: 用於請求主體驗證、回應序列化和清晰的資料結構。這取代了 PHP 中的手動 isset 和 is_int 檢查。
依賴注入 (Depends): 用於管理資料庫連線，確保每個請求都能正確開啟和關閉。
路徑參數: /api/orders/{order_id} 中的 order_id。
查詢參數: 用於分頁的 page 和 limit。
錯誤處理: HTTPException 用於拋出具有適當狀態碼和詳細資訊的特定 HTTP 錯誤。這取代了 PHP 中手動的 http_response_code 和 json_encode 錯誤處理。
自動文件: FastAPI 會根據您的程式碼和 Pydantic 模型自動生成 OpenAPI (Swagger UI) 和 ReDoc 文件。
非同步端點 (async def): 雖然 mysql.connector 是同步的，但 FastAPI 可以很好地處理非同步操作。對於真正的非同步資料庫操作，您可以整合非同步 ORM (如 SQLAlchemy 2.0 與 asyncio) 或 databases 函式庫。

考慮事項和改進:

非同步資料庫操作: 對於高並發應用程式，使用非同步資料庫驅動程式 (例如 aiomysql 或 databases 函式庫與 SQLAlchemy Core) 會比同步的 mysql.connector 表現更好。這將允許您的 FastAPI 應用程式在不阻塞資料庫 I/O 的情況下同時處理許多請求。
安全性:
- 環境變數: 雖然已使用，但考慮在生產環境中使用更穩健的機密管理解決方案 (例如 Kubernetes Secrets, Vault)。
- 存取控制 (Access-Control-Allow-Origin): PHP 中的 Access-Control-Allow-Origin: * 用於開發。在生產環境中，應將其限制為特定的網域。FastAPI 的 CORS_Middleware 可以處理此問題。
- 輸入清理/驗證: Pydantic 處理了大部分問題，但對於直接的 SQL 查詢，請確保正確的參數化 (這由 mysql.connector 預設透過 %s 完成)。
日誌記錄: 實作更結構化的日誌記錄，用於錯誤和應用程式事件。
測試: 為您的 API 端點編寫單元測試和整合測試。
訂單過濾: 如需求所述，向 GET /api/orders 端點添加 status 和 date_range 的過濾器將是一個有價值的補充。您可以在函式簽名中添加可選的查詢參數，並動態修改 SQL 查詢。
全域錯誤處理: FastAPI 允許自訂異常處理程式來捕獲特定類型的異常並回傳一致的錯誤回應。
Docker 化: 使用 Docker 將應用程式容器化將簡化部署。

這段程式碼是一個使用 FastAPI 框架和 MySQL 資料庫的電子商務訂單管理 API。它功能完整，結構清晰，非常適合初學者學習 Python 後端開發。

1. 程式碼的整體目的

這個 後端 API 旨在管理電子商務平台的訂單和產品，提供以下核心功能：

建立新訂單：允許使用者購買商品並記錄訂單資訊。
查看訂單：查詢訂單列表或單一訂單的詳細資訊。
更新訂單狀態：修改訂單的處理狀態，例如從「待處理」改為「已送達」。
查看訂單統計：獲取今日訂單數量和總金額等數據。
查詢產品列表：顯示所有未被刪除的產品。

此 API 採用 FastAPI 處理 HTTP 請求，並使用 MySQL 資料庫進行數據儲存。

2. 程式碼結構

整個程式碼主要分為以下幾個部分：

導入模組：載入必要的 Python 函式庫。
環境設定：讀取環境變數並初始化 FastAPI 應用。
資料庫連線：定義連接 MySQL 資料庫的方式。
資料模型：使用 Pydantic 定義 API 的輸入與輸出格式。
API 路由：定義各種 API 功能，例如查詢、建立、更新訂單等。

接下來，我們將逐一解析這些部分。

3. 詳細解析

3.1 導入模組

Python
import os
from datetime import datetime
from math import ceil
from typing import List, Dict, Union, Any
from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import mysql.connector
from mysql.connector import Error

這部分是程式碼的「工具箱」，每個工具都有特定用途：

os: 用於讀取系統環境變數或生成隨機值（例如訂單編號）。
datetime: 處理日期和時間，例如記錄訂單創建時間。
math.ceil: 用於計算分頁的總頁數，確保即使有零頭資料也能顯示完整頁面。
typing: 幫助定義變數的資料類型，讓程式碼更清晰、更易於維護和除錯。
fastapi: 核心網頁框架，負責處理所有的 HTTP 請求（如 GET、POST 等）。其中：
- HTTPException: 用於回傳標準的 HTTP 錯誤訊息。
- status: 提供各種 HTTP 狀態碼，讓回傳結果更具語義。
- Depends: 用於管理相依性注入，例如自動處理資料庫連線。
pydantic: 強大的資料驗證庫，用於定義資料結構和驗證輸入資料的正確性。
dotenv: 從 .env 檔案載入環境變數，避免將敏感資訊（如資料庫密碼）直接寫在程式碼中。
mysql.connector: Python 連接 MySQL 資料庫的官方驅動，用於執行資料庫操作。
mysql.connector.Error: 處理 MySQL 連線或操作時可能發生的錯誤。

為什麼這樣寫？

這些模組是開發高效、安全且可維護的後端 API 的標準實踐。FastAPI 負責網頁服務，Pydantic 確保資料品質，MySQL 處理數據持久化，而 dotenv 則提升了安全性。對於初學者而言，可以將其視為搭建後端服務的必備「積木」。

3.2 環境設定

Python
load_dotenv()

app = FastAPI(
    title="電子商務訂單管理 API",
    description="用於管理訂單、產品和訂單統計的 API。",
    version="1.0.0"
)

這部分是 FastAPI 應用程式的「啟動設定」：

load_dotenv(): 程式啟動時，它會自動讀取專案根目錄下 .env 檔案中的環境變數。例如，DB_HOST=localhost、DB_NAME=ecommerce_test、DB_USER=root、DB_PASSWORD=your_password 等。這樣做可以避免將敏感資訊硬編碼在程式碼中，大幅提升安全性與部署靈活性。
app = FastAPI(...): 建立一個 FastAPI 應用程式實例。其中設定的 title、description 和 version 等元數據，會自動生成並顯示在 API 的互動式文件（Swagger UI 或 ReDoc）中，方便開發者和使用者了解 API 的功能和用途。

為什麼這樣寫？

這是一個標準的 FastAPI 初始化流程。使用環境變數是業界推薦的配置方式，有助於區分不同環境（開發、測試、生產）的配置，同時保護敏感資訊。FastAPI 的元數據則極大地提升了 API 的可發現性和易用性。

3.3 資料庫連線

Python
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "ecommerce_test")
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")

def get_db_connection():
    conn = None
    try:
        conn = mysql.connector.connect(
            host=DB_HOST,
            database=DB_NAME,
            user=DB_USER,
            password=DB_PASSWORD
        )
        if conn.is_connected():
            print("成功連接到資料庫")
        return conn
    except Error as e:
        print(f"連接 MySQL 資料庫時發生錯誤: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="伺服器錯誤: 無法連接資料庫"
        )

def get_db():
    db = get_db_connection()
    try:
        yield db
    finally:
        if db:
            db.close()

這部分負責建立和管理與 MySQL 資料庫的連接：

環境變數: os.getenv 會從前面載入的環境變數中獲取資料庫連線資訊，如果環境變數不存在，則使用預設值（例如 localhost）。
get_db_connection(): 這個函數負責建立實際的 MySQL 連線。它嘗試連接資料庫，如果成功會列印「成功連接到資料庫」，並回傳連線物件。如果連接失敗，則會捕獲 mysql.connector.Error，並拋出一個 HTTPException，回傳 500 內部伺服器錯誤給客戶端，表示資料庫連線出現問題。
get_db(): 這是一個 FastAPI 特有的 依賴注入 函數。它使用 yield 關鍵字，這意味著它是一個生成器。
- 當任何 API 路由需要資料庫連線時，FastAPI 會呼叫 get_db()。
- get_db() 會先透過 get_db_connection() 取得一個資料庫連線。
- yield db 會將這個連線 db 提供給 API 路由使用。
- 一旦 API 路由的請求處理完畢，無論成功或失敗（即使發生錯誤），finally 區塊都會被執行，確保資料庫連線 db.close() 被安全關閉，避免資源洩漏。

為什麼這樣寫？

這種模式確保了資料庫連線的可靠性和高效性：

安全性: 使用環境變數而非硬編碼。
錯誤處理: 即時回報連線問題。
資源管理: get_db 結合 yield 和 finally，是 FastAPI 中管理資料庫連線的標準且推薦做法，它保證了每個請求都能獲得一個獨立的資料庫連線，並且在請求結束後自動關閉，避免了手動管理連線的複雜性與潛在問題。對於初學者，可以將 get_db 理解為一個「自動借還」資料庫連線的服務。

3.4 資料模型（Pydantic）

Python
class OrderItemRequest(BaseModel):
    product_id: int = Field(..., gt=0, description="產品 ID")
    quantity: int = Field(..., gt=0, description="產品數量")

class CreateOrderRequest(BaseModel):
    user_id: int = Field(..., gt=0, description="下訂單的使用者 ID")
    items: List[OrderItemRequest] = Field(..., min_items=1, description="訂單中的商品清單")

class OrderResponse(BaseModel):
    id: int
    user_id: int
    number: str
    status: str
    total_amount: float
    created_at: datetime

這部分使用 Pydantic 定義了 API 的資料模型，就像為數據建立了「藍圖」和「驗證規則」。

BaseModel: 這是 Pydantic 模型的基類，所有資料模型都需要繼承它。
OrderItemRequest: 定義了單一訂單商品的請求格式。
- product_id: int = Field(..., gt=0, description="產品 ID"): 必須是整數，且值必須大於 0 (gt=0)。... 表示這個欄位是必填的，description 會顯示在 API 文件中。
- quantity: int = Field(..., gt=0, description="產品數量"): 必須是整數，且值必須大於 0。
CreateOrderRequest: 定義了建立新訂單的請求格式。
- user_id: int = Field(..., gt=0, description="下訂單的使用者 ID"): 使用者 ID 必須為大於 0 的整數。
- items: List[OrderItemRequest] = Field(..., min_items=1, description="訂單中的商品清單"): 訂單中的商品清單，必須是一個包含 OrderItemRequest 物件的列表，且至少要包含一個商品 (min_items=1)。
OrderResponse: 定義了回傳給客戶端的訂單資料格式。FastAPI 會自動將資料庫查詢結果轉換成這個模型指定的格式。

為什麼這樣寫？

Pydantic 在後端開發中扮演著「資料守門員」的角色：

自動資料驗證: 它會自動檢查傳入 API 的數據是否符合模型定義的類型和規則（例如 gt=0、min_items=1）。如果數據不符合，會自動回傳清晰的錯誤訊息給客戶端，省去了大量手動驗證的程式碼。
資料轉換: Pydantic 會自動將輸入的資料轉換為正確的 Python 數據類型（如 int、float、datetime）。
自動生成文件: 這些模型定義會被 FastAPI 用來自動生成互動式的 API 文件（Swagger UI），讓其他開發者能清晰地了解 API 的請求和回應格式。

對於初學者，可以把 Pydantic 模型想像成「表單的規則」，它確保了所有提交的資料都是合法的，並且格式正確。

3.5 API 路由

這是程式的核心部分，定義了 API 的各種功能。每個路由都對應一個特定的 HTTP 方法（GET, POST, PUT 等）和一個 URL 路徑。

3.5.1 獲取訂單列表

Python
@app.get("/api/orders", response_model=OrderListResponse)
async def get_orders(page: int = 1, limit: int = 20, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    if limit <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="限制必須是正整數。")
    if page <= 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="頁碼必須是正整數。")

    offset = (page - 1) * limit
    cursor = db.cursor(dictionary=True)

    try:
        cursor.execute("SELECT COUNT(*) FROM orders")
        total_items = cursor.fetchone()['COUNT(*)']
        total_pages = ceil(total_items / limit) if total_items > 0 else 0

        cursor.execute(
            "SELECT id, user_id, number, status, total_amount, created_at FROM orders LIMIT %s OFFSET %s",
            (limit, offset)
        )
        orders = cursor.fetchall()

        return {
            "data": orders,
            "page": page,
            "total_pages": total_pages,
            "total_items": total_items
        }
    except Error as e:
        print(f"資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="伺服器錯誤: 無法獲取訂單列表")
    finally:
        cursor.close()

功能: 處理客戶端發送的 GET 請求，用於查詢訂單列表，並支援分頁功能。
@app.get("/api/orders"): 這個裝飾器 (decorator) 將 get_orders 函數註冊為處理 /api/orders 路徑上 GET 請求的處理器。response_model=OrderListResponse 則指定了回傳資料的 Pydantic 模型，FastAPI 會自動驗證並序列化回傳數據。
參數:
- page: int = 1: 客戶端可以透過 URL 參數指定頁碼，預設為第一頁。
- limit: int = 20: 客戶端可以指定每頁顯示的資料筆數，預設為 20 筆。
- db: mysql.connector.connection.MySQLConnection = Depends(get_db): 這就是前面提到的依賴注入。當請求到達這個路由時，get_db() 函數會被自動呼叫，並將資料庫連線物件 (db) 傳入 get_orders 函數。
邏輯流程:
1. 參數驗證: 首先檢查 limit 和 page 是否為正整數，如果不是則回傳 400 Bad Request 錯誤。
2. 計算偏移量 (offset): 根據 page 和 limit 計算出從資料庫哪一行開始取數據。
3. 獲取總筆數: 執行 SELECT COUNT(*) 查詢資料庫中所有訂單的總數，並計算出總頁數 (total_pages)。
4. 查詢訂單數據: 執行 SELECT ... LIMIT %s OFFSET %s 查詢，這是一個標準的分頁 SQL 語句。%s 是參數化查詢，可以有效防止 SQL 注入攻擊。
5. 回傳結果: 將查詢到的訂單數據、當前頁碼、總頁數和總筆數包裝成字典回傳。
6. 錯誤處理 (try...except...finally):
  - try 區塊包含所有可能發生資料庫錯誤的程式碼。
  - 如果發生 mysql.connector.Error，則捕獲錯誤並回傳 500 Internal Server Error。
  - finally 區塊確保無論成功或失敗，cursor.close() 都會被執行，釋放資料庫游標資源。

為什麼這樣寫？

這是處理大量數據並提供友善使用者體驗的標準方法——分頁查詢。它避免了一次性載入所有數據導致伺服器壓力過大。try-except-finally 結構確保了錯誤處理的健壯性，而參數化查詢則是防止 SQL 注入的關鍵。對於初學者，可以將這部分想像成「從倉庫中每次只拿一小批商品，並告訴客戶總共有多少批」。

3.5.2 建立新訂單

Python
@app.post("/api/orders", status_code=status.HTTP_201_CREATED)
async def create_order(order_data: CreateOrderRequest, db: mysql.connector.connection.MySQLConnection = Depends(get_db)):
    cursor = db.cursor()
    try:
        db.start_transaction()
        user_id = order_data.user_id
        order_number = f"ORD{datetime.now().strftime('%Y%m%d%H%M%S')}{os.urandom(2).hex()}"
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

        cursor.execute(
            "INSERT INTO orders (user_id, number, status, created_at, updated_at) VALUES (%s, %s, 'pending', %s, %s)",
            (user_id, order_number, current_time, current_time)
        )
        order_id = cursor.lastrowid
        total_amount = 0.0

        for item in order_data.items:
            product_id = item.product_id
            quantity = item.quantity
            cursor.execute("SELECT stock, price, is_deleted FROM products WHERE id = %s FOR UPDATE", (product_id,))
            product = cursor.fetchone()

            if not product:
                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"產品 ID {product_id} 不存在")
            
            product_stock, product_price, is_deleted = product[0], product[1], product[2]
            if is_deleted:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"產品 ID {product_id} 已停用或刪除")
            if product_stock < quantity:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"產品 ID {product_id} 庫存不足。可用: {product_stock}, 需求: {quantity}")

            unit_price = float(product_price)
            subtotal = unit_price * quantity
            total_amount += subtotal

            cursor.execute(
                "UPDATE products SET stock = stock - %s, updated_at = %s WHERE id = %s AND stock >= %s",
                (quantity, current_time, product_id, quantity)
            )
            cursor.execute(
                "INSERT INTO order_items (order_id, product_id, quantity, unit_price, subtotal, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s)",
                (order_id, product_id, quantity, unit_price, subtotal, current_time, current_time)
            )

        cursor.execute("UPDATE orders SET total_amount = %s WHERE id = %s", (total_amount, order_id))
        db.commit()
        return {"order_id": order_id, "order_number": order_number}
    except HTTPException as e:
        db.rollback()
        raise e
    except Error as e:
        db.rollback()
        print(f"訂單創建期間的資料庫錯誤: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"伺服器錯誤: 無法創建訂單")
    finally:
        cursor.close()

功能: 處理客戶端發送的 POST 請求，用於建立新的訂單，並自動處理庫存扣減。
@app.post("/api/orders", status_code=status.HTTP_201_CREATED): 將 create_order 函數註冊為處理 /api/orders 路徑上 POST 請求的處理器。status_code=status.HTTP_201_CREATED 表示成功建立資源時回傳 201 Created 狀態碼。
參數: order_data: CreateOrderRequest，FastAPI 會自動驗證請求體（request body）中的 JSON 資料是否符合 CreateOrderRequest Pydantic 模型的定義。
邏輯流程:
1. 開啟資料庫事務 (db.start_transaction()): 這是建立訂單最關鍵的一步。事務確保了在一個「邏輯單元」內的所有資料庫操作要么全部成功提交 (db.commit())，要么全部失敗回滾 (db.rollback())。這避免了只扣了庫存卻沒建立訂單，或反之，確保了資料的原子性和一致性。
2. 生成訂單編號: 透過當前時間和隨機數生成一個唯一的訂單編號（例如 ORD20250612195812a3b4）。
3. 插入主訂單: 將使用者 ID、生成的訂單編號和狀態（預設為 'pending'）插入到 orders 表中。cursor.lastrowid 用於獲取新插入訂單的 ID。
4. 處理訂單商品: 遍歷 order_data.items 中的每個商品：
  - 查詢產品資訊並鎖定 (FOR UPDATE): 執行 SELECT stock, price, is_deleted FROM products WHERE id = %s FOR UPDATE。FOR UPDATE 是 MySQL 的行級鎖，它會在查詢的瞬間鎖定這行產品數據，防止其他並發請求同時修改庫存，這對於電商的庫存管理至關重要，防止超賣。
  - 庫存和狀態檢查: 檢查產品是否存在、是否已刪除或停用，以及庫存是否充足。如果條件不滿足，則拋出 HTTPException 回傳對應的錯誤碼和訊息。
  - 計算小計: 計算單個商品的總價 (unit_price * quantity)。
  - 更新產品庫存: 將產品的庫存減少 (stock = stock - %s)。
  - 插入訂單商品: 將商品資訊插入到 order_items 表中。
5. 更新訂單總金額: 計算所有商品的小計總和，並更新 orders 表中的 total_amount。
6. 提交事務 (db.commit()): 如果所有操作都成功，則提交整個事務，將更改永久保存到資料庫。
7. 錯誤處理:
  - 如果執行過程中拋出任何 HTTPException（例如產品不存在、庫存不足），會立即進入 except HTTPException 區塊，執行 db.rollback() 撤銷所有之前的資料庫操作，然後重新拋出這個錯誤。
  - 如果發生其他資料庫錯誤（Error as e），同樣執行 db.rollback() 並回傳 500 錯誤。
  - finally 區塊確保 cursor.close() 被執行。

為什麼這樣寫？

這是電子商務中最核心且複雜的功能之一。它示範了如何處理：

資料庫事務: 確保操作的原子性和資料一致性。
並發控制: 使用 FOR UPDATE 鎖定機制來防止高並發下的超賣問題。
詳細的業務邏輯驗證: 在資料庫操作前和操作中進行多重檢查，確保業務規則的正確執行。

對於初學者，可以把這個流程想像成「一個複雜的購物結帳流程」，每一步都要確認無誤，才能最終確認訂單，如果中間有任何問題，所有操作都將取消。

3.5.3 其他路由

這段程式碼還包含了其他重要的 API 路由，它們的結構與上述例子類似，主要差異在於它們執行的 SQL 查詢和回傳的數據內容：

更新訂單狀態 (PUT /api/orders/{order_id}/status)：
- 接收訂單 ID 和新的狀態。
- 執行一個簡單的 SQL UPDATE 語句來修改訂單狀態。
- 通常會包含權限檢查（雖然此處未示範，但實際應用中應考慮）。
獲取訂單統計 (GET /api/orders/stats)：
- 查詢資料庫，使用 SQL 聚合函數（如 COUNT()、SUM()）來計算總訂單數、總金額、今日訂單數和金額等統計數據。
- 回傳這些統計結果。
獲取產品列表 (GET /api/products)：
- 與獲取訂單列表類似，支援分頁。
- 查詢 products 表，只回傳 is_deleted 為 0（未刪除）的產品。

這些路由共同構成了這個電子商務訂單管理 API 的完整功能集。

4. 程式碼的寫法和邏輯

這段程式碼的寫法遵循了許多現代後端開發的最佳實踐，非常適合初學者學習：

模組化設計：
- 將不同的功能邏輯分離到不同的模組或函數中，例如資料庫連線 (get_db)、資料驗證 (Pydantic 模型)、API 邏輯 (各個路由函數)。
- 每個函數或類別都承擔單一職責，使程式碼更易於理解、測試和維護。
全面錯誤處理：
- 廣泛使用 try/except 區塊來捕獲和處理資料庫操作、輸入驗證等可能發生的錯誤。
- 回傳清晰且具語義的 HTTP 錯誤狀態碼（如 400 Bad Request, 404 Not Found, 500 Internal Server Error），方便客戶端調試和響應。
- 資料庫事務管理 (db.start_transaction(), db.commit(), db.rollback()) 確保了多個相關資料庫操作的原子性，避免資料不一致。
安全性考量：
- 環境變數 (.env) 保護了資料庫憑證等敏感資訊，避免硬編碼。
- Pydantic 資料驗證 自動檢查 API 輸入的數據類型、格式和業務規則，有效防止了無效數據和潛在的惡意輸入。
- 參數化 SQL 查詢 (%s) 徹底杜絕了 SQL 注入攻擊，這在任何與資料庫交互的應用中都是至關重要的。
高可讀性與可維護性：
- 使用 型別提示 (List[OrderItemRequest]) 讓程式碼的意圖更加明確，工具可以提供更好的自動完成和靜態分析。
- 清晰的變數命名（例如 total_amount、order_id）使程式碼自解釋性強。
- FastAPI 裝飾器中的 summary 和 description 等參數有助於自動生成清晰的 API 文件。
效率和資源管理：
- 分頁查詢 (LIMIT 和 OFFSET) 避免了一次性載入大量數據，優化了效能。
- 資料庫連線在請求結束後自動關閉 (finally 區塊中的 db.close())，有效管理了伺服器資源。
- FOR UPDATE 的使用有效地處理了並發環境下的庫存問題。

初學者怎麼理解？

想像這段程式碼就是一個高效運作的「線上商店後台經理」。當顧客發送一個請求（例如下訂單），這個經理會：

FastAPI：是前台接待員，接收請求。
Pydantic：是嚴格的質檢員，檢查顧客的訂單資訊是否合規。
MySQL：是龐大的倉庫，儲存著商品和訂單的詳細記錄。
get_db：是倉庫管理員，每次客戶需要操作倉庫時，他都會提供一個專屬的「通道」，用完後會立即關閉。
Transaction：是簽訂的「合同」，確保所有操作（例如扣庫存和建立訂單）要么全部完成，要么全部取消。
FOR UPDATE：是倉庫經理對特定商品的「臨時封鎖」，確保在處理當前訂單時，沒有其他客戶能同時拿走這些商品。

5. 初學者如何學習這段程式碼

要有效地學習這段程式碼，建議從以下幾個步驟入手：

先讓程式跑起來
- 安裝依賴: 打開終端機，執行 pip install fastapi uvicorn pymysql python-dotenv。
- 建立 MySQL 資料庫: 根據程式碼中的 DB_NAME（例如 ecommerce_test），在你的 MySQL 伺服器中建立一個資料庫。
- 建立資料表: 根據程式碼邏輯，手動或使用 SQL 腳本建立 orders、products 和 order_items 表。確保欄位名稱和類型與程式碼中的查詢匹配。
  - products 表: 至少包含 id, stock, price, is_deleted 等欄位。
  - orders 表: 至少包含 id, user_id, number, status, total_amount, created_at, updated_at 等欄位。
  - order_items 表: 至少包含 id, order_id, product_id, quantity, unit_price, subtotal, created_at, updated_at 等欄位。
  - 可以插入一些測試數據到 products 表中，以便後續測試訂單創建功能。
- 建立 .env 檔案: 在你的程式碼根目錄下建立一個名為 .env 的檔案，並填入你的 MySQL 連線資訊：
  DB_HOST=your_mysql_host # 通常是 localhost 或 127.0.0.1 DB_NAME=ecommerce_test # 你的資料庫名稱 DB_USER=your_mysql_user DB_PASSWORD=your_mysql_password
- 執行程式: 在終端機中，切換到你的程式碼目錄，執行 uvicorn main:app --reload (假設你的 FastApi 程式碼儲存在 main.py 中，且 app 是 FastAPI 實例)。
- 訪問 API 文件: 開啟瀏覽器，訪問 http://localhost:8000/docs。你將看到自動生成的互動式 API 文件（Swagger UI），你可以在這裡直接測試各個 API 接口。
分段學習，逐個擊破
- 從 GET 路由開始: 先理解像 /api/orders 這樣簡單的 GET 請求如何查詢資料並回傳。這會讓你熟悉 FastAPI 的基本結構和資料庫查詢。
- 深入 POST 路由: 接著研究 /api/orders 的 POST 請求，這是最複雜的部分，可以幫助你理解資料庫事務、並發處理和多步驟邏輯。
- 研究 Pydantic 模型: 學習 OrderItemRequest、CreateOrderRequest 等模型如何定義輸入和輸出的數據結構，以及 Pydantic 如何進行資料驗證。
- 理解資料庫連線和事務: 深入理解 get_db_connection() 和 get_db() 函數，以及在 create_order 路由中 start_transaction() 和 commit()/rollback() 的作用。
動手修改，加深理解
- 新增簡單 API: 試著新增一個簡單的 API 路由，例如：
  - GET /api/users/{user_id}/orders: 查詢某個特定用戶的所有訂單。
  - POST /api/products: 新增一個產品到資料庫。
- 修改錯誤訊息: 嘗試修改某些 HTTPException 的 detail 訊息，使其更符合你的需求。
- 增加驗證規則: 在 Pydantic 模型中增加新的 Field 驗證，例如限制產品名稱的長度，或訂單總金額的範圍。
利用學習資源
- FastAPI 官方教程: 這是最好的學習資源，官方文件非常詳細且易懂：https://fastapi.tiangolo.com/
- Pydantic 入門: 深入了解 Pydantic 的資料驗證和模型定義：https://pydantic-docs.helpmanual.io/
- MySQL 基礎: 學習基本的 SQL 語句（SELECT, INSERT, UPDATE, DELETE）和資料庫概念。
- Python 型別提示: 理解 List、Dict、Union 等型別提示的使用，這會讓你的程式碼更規範。

6. 總結

這段程式碼提供了一個完整的電子商務後端 API 範例，清晰展示了如何結合 FastAPI、Pydantic 和 MySQL 來管理數據。其專業的寫法和實用的功能使其成為學習 Python 後端開發的絕佳教材。

核心要點概括:

FastAPI: 快速構建高性能 API 的框架。
Pydantic: 強大的資料驗證和模型定義工具。
MySQL: 負責資料的持久化儲存，並透過事務保證資料一致性。
結構清晰: 程式碼模組化、注重錯誤處理、兼顧安全性和執行效率。

透過理解和實踐這段程式碼，你將能掌握 Python 後端開發的核心概念和實戰技巧。

專案 GitHub 連結： https://github.com/BpsEason/ecommerce_fastapi.git
感謝你的閱讀！希望這篇文章能幫助你打開高併發爬蟲世界的大門。

程式解惑室：羅煜鈞的開發日誌與技術分享

2025年6月12日星期四

FastAPI 電商訂單系統：高效能 RESTful API 實戰

這段程式碼是一個使用 FastAPI 框架和 MySQL 資料庫的電子商務訂單管理 API。它功能完整，結構清晰，非常適合初學者學習 Python 後端開發。

1. 程式碼的整體目的

2. 程式碼結構

3. 詳細解析

3.1 導入模組

3.2 環境設定

3.3 資料庫連線

3.4 資料模型（Pydantic）

3.5 API 路由

3.5.1 獲取訂單列表

3.5.2 建立新訂單

3.5.3 其他路由

4. 程式碼的寫法和邏輯

5. 初學者如何學習這段程式碼

先讓程式跑起來

分段學習，逐個擊破

動手修改，加深理解

利用學習資源

6. 總結

沒有留言:

張貼留言

Laravel 進階開發：從多租戶 SaaS 專案學習中階實踐

網誌存檔

搜尋此網誌

2025年6月12日 星期四

FastAPI 電商訂單系統：高效能 RESTful API 實戰

這段程式碼是一個使用 FastAPI 框架和 MySQL 資料庫的電子商務訂單管理 API。它功能完整，結構清晰，非常適合初學者學習 Python 後端開發。

1. 程式碼的整體目的

2. 程式碼結構

3. 詳細解析

3.1 導入模組

3.2 環境設定

3.3 資料庫連線

3.4 資料模型（Pydantic）

3.5 API 路由

3.5.1 獲取訂單列表

3.5.2 建立新訂單

3.5.3 其他路由

4. 程式碼的寫法和邏輯

5. 初學者如何學習這段程式碼

先讓程式跑起來

分段學習，逐個擊破

動手修改，加深理解

利用學習資源

6. 總結

沒有留言:

張貼留言

Laravel 進階開發：從多租戶 SaaS 專案學習中階實踐

網誌存檔

2025年6月12日星期四