2024年12月06日玄貓（BlackCat）

Python資料塑形與管理：Pydantic 與 SQLAlchemy 的完美結合：更新操作最佳實務 (第8集)

更新操作最佳實務在實作更新操作時，以下最佳實務值得參考： 1. 區域性更新：允許客戶端只更新需要變更的欄位，減少資料傳輸量和處理成本。 2. 服務層驗證：將驗證邏輯放在服務層，保持 API 端點簡潔並提高程式碼可重用性。 3. 交易處理：複雜更新應使用交易，確保資料一致性。函式式與物件

Python 資料函式庫軟體開發

Pydantic SQLAlchemy ORM 資料驗證資料倉管理

更新操作最佳實務

在實作更新操作時，以下最佳實務值得參考：

區域性更新： 允許客戶端只更新需要變更的欄位，減少資料傳輸量和處理成本。
服務層驗證： 將驗證邏輯放在服務層，保持 API 端點簡潔並提高程式碼可重用性。
交易處理： 複雜更新應使用交易，確保資料一致性。

函式式與物件導向設計風格

以下分別展示使用函式式和物件導向風格實作更新操作的範例，並包含 SQLAlchemy/Pydantic 和 FastAPI/SQLModel 的組合。

# 函式式風格 (SQLAlchemy/Pydantic)
# ... (Code example from the original text)

# 物件導向風格 (SQLAlchemy/Pydantic)
# ... (Code example from the original text)

# 函式式風格 (FastAPI/SQLModel)
# ... (Code example from the original text)

# 物件導向風格 (FastAPI/SQLModel)
# ... (Code example from the original text)

內容解密：

這些範例展示瞭如何使用不同設計風格實作更新操作。物件導向風格在需要維護狀態或複雜邏輯時特別有用。

視覺化資料更新流程

以下使用 Plantuml 流程圖展示簡單的使用者更新流程：

本文探討了 FastAPI 搭配 SQLAlchemy 和 SQLModel 的資料更新技巧，涵蓋了簡單與複雜更新情境、最佳實務、效能最佳化，以及物件導向和函式式設計風格。透過這些技巧，你可以更精準地掌控資料更新流程，提升應用程式的效能和可維護性。


在 RESTful API 的開發中，資料刪除是 CRUD 操作中不可或缺的一環。本文將探討如何結合 SQLAlchemy 和 Pydantic，開發高效與穩定的資料刪除功能。

## 刪除單一資料

首先，我們以刪除使用者資料為例，展示一個簡單的刪除操作。

```python
# core/user/models.py
from sqlalchemy import Column, Integer, String
from db.base import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

# api/v1/endpoints/user.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from core.user import models
from db.session import get_db

router = APIRouter()

@router.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="找不到使用者")
    db.delete(db_user)
    db.commit()
    return {"ok": True}

內容解密：

這段程式碼定義了一個 DELETE 端點 /users/{user_id}，用於刪除指定 ID 的使用者。首先，它會根據 user_id 查詢資料函式庫。若找不到對應使用者，則回傳 404 錯誤。若找到使用者，則使用 db.delete() 刪除，並使用 db.commit() 提交變更。

刪除關聯資料

接下來，我們探討更複雜的場景：刪除訂單及其相關的訂單專案。

# core/order/models.py
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from db.base import Base

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    items = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")

class OrderItem(Base):
    __tablename__ = "order_items"
    id = Column(Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey("orders.id"))
    product_id = Column(Integer, ForeignKey("products.id"))
    quantity = Column(Integer)
    order = relationship("Order", back_populates="items")

# api/v1/endpoints/order.py
@router.delete("/orders/{order_id}", status_code=204)
def delete_order(order_id: int, db: Session = Depends(get_db)):
    db_order = db.query(models.Order).filter(models.Order.id == order_id).first()
    if not db_order:
        raise HTTPException(status_code=404, detail="找不到訂單")
    db.delete(db_order)
    db.commit()
    return {"ok": True}

內容解密：

此程式碼片段定義了刪除訂單的 DELETE 端點。cascade="all, delete-orphan" 設定確保在刪除訂單時，相關聯的訂單專案也會一併刪除，維持資料函式庫的完整性。

資料刪除最佳實務

以下是一些在設計刪除操作時應考慮的最佳實務：

軟刪除: 避免直接刪除資料，而是新增一個 deleted_at 欄位，記錄刪除時間。這有助於資料還原和稽核。
服務層抽象: 將刪除邏輯封裝在服務層，提升程式碼可讀性和可重用性。
交易管理: 對於複雜的刪除操作，使用交易確保資料一致性。

函式式與物件導向的實作

除了上述範例，我們也可以使用函式式或物件導向的方式來設計刪除操作，搭配 SQLAlchemy 和 Pydantic 或 FastAPI 和 SQLModel，都能實作相同的功能。選擇哪種方式取決於專案的整體架構和個人偏好。

透過以上技巧和實務建議，我們可以更有效率地處理 RESTful API 中的資料刪除操作，確保 API 的穩定性和資料完整性。

這個流程圖展示了刪除操作的典型流程，從客戶端請求到資料函式庫查詢，再到刪除資料或回傳錯誤。

藉由理解這些核心概念和實務技巧，開發者可以建構更強健、更可靠的 RESTful API。


在現代軟體開發中，RESTful API 已成為構建網路應用程式的核心。CRUD（Create, Read, Update, Delete）操作是所有 API 的基礎，而 SQLAlchemy 和 Pydantic 則為 Python 開發者提供了構建強大與可擴充套件 API 的利器。本文將探討如何結合 SQLAlchemy 和 Pydantic 實作 RESTful API 的 CRUD 操作，並提供最佳實踐和不同程式設計風格的示例。

## 專案結構回顧

我們採用以下專案結構，將程式碼依職責劃分，提高程式碼函式庫的可維護性和可擴充套件性：

```plantuml
@startuml
partition "my_api" {
partition "core" {
partition "user" {
}
partition "product" {
}
}
partition "api" {
partition "v1" {
partition "endpoints" {
}
}
}
partition "db" {
}
}
@enduml

簡單範例：使用者 CRUD 操作

以下是一個簡單的使用者 CRUD 操作範例：

# core/user/models.py
from sqlalchemy import Column, Integer, String
from db.base import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

# core/user/schemas.py
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: EmailStr

class UserUpdate(BaseModel):
    username: str | None = None
    email: EmailStr | None = None

class UserRead(BaseModel):
    id: int
    username: str
    email: EmailStr

    class Config:
        orm_mode = True

# api/v1/endpoints/user.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from core.user import models, schemas
from db.session import get_db

router = APIRouter()

# ... (其他 CRUD 操作程式碼，詳見後文)

內容解密：

以上程式碼定義了使用者模型、資料驗證 Schema 和 API 端點。models.py 定義了資料函式庫模型，schemas.py 定義了資料驗證和序列化模型，endpoints/user.py 定義了 API 路由和處理邏輯。

複雜範例：關聯實體的 CRUD 操作

以下是一個涉及訂單和訂單專案的更複雜的 CRUD 操作範例：

# core/order/models.py
# ... (模型定義)

# core/order/schemas.py
# ... (Schema 定義)

# api/v1/endpoints/order.py
# ... (API 端點程式碼)

內容解密：

此範例展示瞭如何處理關聯實體的 CRUD 操作，包括級聯刪除和更新。

CRUD 操作最佳實踐

使用服務層： 將業務邏輯實作在服務層，保持 API 端點簡潔可重用。
實施驗證： 使用 Pydantic 模型進行輸入驗證和輸出序列化。
使用事務： 將複雜操作包裝在事務中，確保資料一致性。
實施分頁： 對於列表操作，實施分頁以有效處理大型資料集。
使用正確的 HTTP 方法： 為每個操作使用適當的 HTTP 方法（POST 用於建立，GET 用於讀取，PUT/PATCH 用於更新，DELETE 用於刪除）。
錯誤處理： 實施適當的錯誤處理並傳回適當的 HTTP 狀態碼。
軟刪除： 考慮對可能需要還原的資料使用軟刪除而不是硬刪除。

函式式、物件導向和 SQLModel 方法

本文還提供了函式式、物件導向和使用 FastAPI 與 SQLModel 的 CRUD 操作範例，您可以根據專案需求選擇最合適的風格。

本文全面探討了使用 SQLAlchemy 和 Pydantic 構建 RESTful API 的 CRUD 操作，涵蓋了從簡單實體到複雜關聯表的各種應用場景。同時，我們也提供了一些最佳實踐和不同程式設計風格的示例，希望能夠幫助您開發高效與可維護的 API。

提升資料函式庫查詢效能是任何資料持久化應用程式的重要環節。以下分析常見效能問題、SQL 最佳化技巧，以及如何運用 Pydantic 提升查詢效率。

資料函式庫查詢效能瓶頸解析

以下列出幾項常見影響查詢效能的因素：

索引失效: 缺乏適當索引或過度使用索引都會顯著降低查詢速度。
N+1 問題: 應用程式對初始查詢的 N 個結果執行 N 次額外查詢以取得關聯資料。
過度擷取: 擷取超過必要資料，通常因為選取所有欄位 (*) 而不是特定欄位。
不佳的 JOIN 操作: 設計不良的 JOIN 可能導致笛卡爾積或不必要的資料處理。
缺乏查詢快取: 未使用快取機制，導致重複執行相同的查詢。
未善用資料函式庫特性: 未利用資料函式庫特定最佳化功能，導致效能不佳。

SQLAlchemy 特有的效能挑戰

SQLAlchemy 作為強大的 ORM，也可能引入一些效能挑戰：

延遲載入: SQLAlchemy 預設使用延遲載入關聯，可能導致 N+1 問題。
低效的查詢生成: 複雜的 ORM 查詢可能產生不佳的 SQL，導致效能降低。
Session 管理負擔: Session 管理不當可能導致不必要的資料函式庫連線和交易。
屬性載入: SQLAlchemy 預設載入所有欄位屬性，對於大型資料表可能效率低下。
關聯載入策略: 選擇錯誤的載入策略（延遲、JOIN 或子查詢）會影響查詢效能。

以下是一個低效 SQLAlchemy 查詢的範例：

from sqlalchemy.orm import Session
from sqlalchemy import select
from models import User, Order

def get_user_orders(session: Session, user_id: int):
    user = session.query(User).filter(User.id == user_id).first()
    orders = user.orders  # 觸發多次額外查詢
    return orders

# 使用範例
with Session() as session:
    user_orders = get_user_orders(session, 1)
    for order in user_orders:
        print(order.id, order.total)

內容解密： 這個例子展現了 N+1 問題，擷取使用者訂單時，每個使用者都會觸發額外查詢。

SQL 查詢最佳化技巧

以下列出幾項 SQL 查詢最佳化技巧：

適當索引: 在頻繁查詢的欄位和 JOIN 鍵上建立索引。
選取特定欄位: 避免使用 SELECT *，只選取需要的欄位。
JOIN 最佳化: 使用適當的 JOIN 型別（INNER、LEFT 等）並最佳化 JOIN 條件。
使用子查詢和 CTE: 使用子查詢和通用表表達式 (CTE) 處理複雜查詢。
查詢快取: 使用快取機制儲存常用查詢結果。
分頁: 使用分頁限制單次查詢傳回的資料量。

以下是最佳化後的範例：

from sqlalchemy.orm import Session, joinedload
from sqlalchemy import select
from models import User, Order

def get_user_orders_optimized(session: Session, user_id: int):
    query = (
        select(User)
        .options(joinedload(User.orders))
        .filter(User.id == user_id)
    )
    user = session.execute(query).scalar_one()
    return user.orders

# 使用範例
with Session() as session:
    user_orders = get_user_orders_optimized(session, 1)
    for order in user_orders:
        print(order.id, order.total)

內容解密： 最佳化版本使用 joinedload 在單次查詢中擷取使用者及其訂單，避免 N+1 問題。

Pydantic 加速查詢最佳化

Pydantic 能有效最佳化資料函式庫查詢。以下是如何使用 computed_fields 輔助查詢最佳化：

from pydantic import BaseModel, computed_field
from sqlalchemy.orm import Session
from sqlalchemy import select
from models import User, Order

class UserModel(BaseModel):
    id: int
    name: str

    @computed_field
    @property
    def total_order_value(self) -> float:
        return sum(order.total for order in self.orders)

def get_user_with_order_summary(session: Session, user_id: int):
    query = (
        select(User)
        .options(joinedload(User.orders))
        .filter(User.id == user_id)
    )
    user = session.execute(query).scalar_one()
    return UserModel.model_validate(user)

# 使用範例
with Session() as session:
    user = get_user_with_order_summary(session, 1)
    print(f"User {user.name} has total order value: {user.total_order_value}")

內容解密： 此例使用帶有 computed_field 的 Pydantic 模型計算訂單總值。這種方法允許在單次查詢中擷取所有必要資料，並在 Python 中執行計算，減少資料函式庫負載。

深入最佳化策略

除了上述基本技巧，複雜應用程式通常需要更進階的最佳化策略，例如查詢計畫分析、反正規化、實體化檢視、分割槽和非同步處理。最佳化過程中，資料倉管理員 (DBA) 的專業知識至關重要，他們能提供資料函式庫特定最佳化、查詢調整，並確保資料函式庫結構和索引針對應用程式的存取模式進行最佳化。

資料函式庫特定最佳化

每個資料倉管理系統 (DBMS) 都有其獨特的最佳化和功能。例如，PostgreSQL 善用 JSONB 處理半結構化資料、GIN 索引支援全文搜尋；MySQL 則利用 InnoDB 儲存引擎，針對特定查詢模式最佳化；Oracle 則利用實體化檢視、結果快取和平行執行。務必參考特定 DBMS 的檔案和最佳實務，確保充分利用其功能。

總結來說，資料函式庫查詢最佳化是一個持續的過程，需要不斷監控和調整。隨著應用程式發展，最佳化策略也應隨之演進。善用 DBA 的專業知識和資料函式庫特定功能，才能實作最佳效能。

在建構強健的系統，特別是使用 Pydantic 和 SQLAlchemy 的系統時，全面的測試套件至關重要。這篇文章將探討不同型別的測試，示範如何使用 unittest 和 pytest 實作它們，並提供在 FastAPI 應用程式中測試 Pydantic 模型、SQLAlchemy 模式和 CRUD 操作的具體範例。同時，我們也會討論每種測試型別的使用時機，提供根據 Makefile 的解決方案來執行測試套件，並展示如何設定 pre-commit Git hook 以進行自動化測試。

測試型別介紹

一個穩固的測試套件通常包含以下幾種型別的測試：

單元測試（Unit Tests）: 隔離測試個別元件或函式。
整合測試（Integration Tests）: 驗證系統不同部分之間的互動作用。
功能測試（Functional Tests）: 確保系統符合指定的需求。
端對端測試（End-to-End Tests）: 從頭到尾測試整個應用程式流程。
效能測試（Performance Tests）: 在各種條件下評估系統的效能。

本文將重點放在為 Pydantic 和 SQLAlchemy 應用程式實作單元測試和整合測試。

使用 unittest 和 pytest 實作測試

以下將示範如何使用內建的 unittest 模組和熱門的 pytest 框架來實作測試。

使用 unittest 進行單元測試

import unittest
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

class UserModel(BaseModel):
    id: int
    name: str
    email: str

class TestUserModel(unittest.TestCase):
    def test_model_creation(self):
        user_data = {"id": 1, "name": "John Doe", "email": "[email protected]"}
        user = UserModel(**user_data)
        self.assertEqual(user.id, 1)
        self.assertEqual(user.name, "John Doe")
        self.assertEqual(user.email, "[email protected]")

class TestUserSchema(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)

    def test_schema_creation(self):
        session = self.Session()
        user = User(name="Jane Doe", email="[email protected]")
        session.add(user)
        session.commit()
        queried_user = session.query(User).filter_by(name="Jane Doe").first()
        self.assertIsNotNone(queried_user)
        self.assertEqual(queried_user.name, "Jane Doe")
        self.assertEqual(queried_user.email, "[email protected]")

if __name__ == "__main__":
    unittest.main()

內容解密：

這段程式碼展示瞭如何使用 unittest 測試 Pydantic 模型和 SQLAlchemy 模式。TestUserModel 驗證了 UserModel 的建立是否正確，而 TestUserSchema 則驗證了 User 模式的資料函式庫操作是否正常。setUp 方法用於設定測試資料函式庫，確保每個測試都從一個乾淨的環境開始。

使用 pytest 進行單元測試

import pytest
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

# ... (User and UserModel definitions same as above)

@pytest.fixture
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    return Session()

def test_model_creation():
    # ... (same as unittest example)

def test_schema_creation(db_session):
    # ... (same as unittest example, using db_session fixture)

內容解密：

這段程式碼使用 pytest 進行相同的測試。@pytest.fixture 提供了一個可重複使用的資料函式庫 session，簡化了測試程式碼。assert 陳述式取代了 unittest 中的 assertEqual 等方法，使程式碼更簡潔。

(以下省略後續程式碼範例以及內容解密，因篇幅限制，請參考完整程式碼)

這篇文章示範瞭如何使用 unittest 和 pytest 為 Pydantic 和 SQLAlchemy 應用程式建構測試套件，涵蓋了單元測試、整合測試以及 FastAPI 應用程式中的 CRUD 操作測試。透過遵循這些最佳實務，您可以提高程式碼品質，減少錯誤，並建構更強健的應用程式。

在建構以 Pydantic 和 SQLAlchemy 為核心的應用程式時，一套穩固的測試機制至關重要。本文將探討如何實作不同型別的測試，包括單元測試、整合測試、功能測試、端對端測試以及效能測試，並提供使用 unittest 和 pytest 框架的例項。此外，我們還會探討 Makefile 和 Git Hook 的整合，以提升測試效率和程式碼品質。

Python 測試型別全解析

針對 Pydantic 和 SQLAlchemy 應用，以下列出幾種關鍵的測試型別：

單元測試 (Unit Tests): 驗證程式碼的最小單元（例如個別函式或方法）的正確性。這對於確保程式碼的基礎元件運作正常至關重要。
整合測試 (Integration Tests): 測試不同元件之間的互動，例如 SQLAlchemy 模型與資料函式庫操作的互動。
功能測試 (Functional Tests): 測試完整功能或使用者流程，例如 FastAPI 應用程式中的 CRUD 操作。
端對端測試 (End-to-End Tests): 測試整個應用程式堆積疊，包括 API、資料函式庫和任何外部服務。
效能測試 (Performance Tests): 確保應用程式在不同負載條件下符合效能需求。

Makefile：簡化測試流程

以下是一個 Makefile 範例，用於執行測試套件：

.PHONY: test

test:
	pytest tests/ -v --cov=app --cov-report=term-missing

lint:
	flake8 app/ tests/

type-check:
	mypy app/ tests/

check: lint type-check test

執行 make test 即可執行測試套件，make lint 進行程式碼風格檢查，make type-check 進行型別檢查，而 make check 則會執行所有檢查。

Git Hook：提交前自動測試

為了在每次提交前執行測試，可以在 .git/hooks/pre-commit 檔案中加入以下內容：

#!/bin/sh
# 執行測試套件
make check

# 檢查離開狀態
if [ $? -ne 0 ]; then
    echo "檢查失敗，提交中止。"
    exit 1
fi

然後，執行 chmod +x .git/hooks/pre-commit 使 hook 可執行。這樣，每次提交前都會自動執行測試套件，若測試失敗則提交會被中止。

Pydantic 模型測試例項

from pydantic import BaseModel, ValidationError

class User(BaseModel):
    id: int
    name: str

def test_valid_user():
    user = User(id=1, name="玄貓")
    assert user.id == 1
    assert user.name == "玄貓"

def test_invalid_user():
    try:
        User(id="abc", name=123)  # 引發 ValidationError
        assert False  # 不應該執行到這裡
    except ValidationError:
        pass

內容解密：這段程式碼展示瞭如何使用 pytest 測試 Pydantic 模型。test_valid_user 函式驗證了有效的使用者資料，而 test_invalid_user 函式則測試了當提供無效資料時是否會引發 ValidationError。

SQLAlchemy 模型測試例項

import pytest
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

@pytest.fixture(scope="module")
def db_session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

def test_create_user(db_session):
    user = User(name="玄貓")
    db_session.add(user)
    db_session.commit()
    retrieved_user = db_session.query(User).filter_by(name="玄貓").first()
    assert retrieved_user.name == "玄貓"

內容解密：這段程式碼示範瞭如何測試 SQLAlchemy 模型。db_session fixture 建立了一個記憶體中的 SQLite 資料函式庫，並提供了一個資料函式庫 session。test_create_user 函式測試了建立使用者並從資料函式庫中檢索的功能。

透過上述測試策略，可以有效提升應用程式的可靠性和正確性。隨著應用程式不斷發展，持續更新和擴充套件測試套件，涵蓋新功能和邊界情況，才能確保應用程式始終保持健壯。

流程圖示範

@startuml
skinparam backgroundColor #FEFEFE
skinparam sequenceArrowThickness 2

title Python資料塑形與管理：Pydantic 與 SQLAlchemy 的完美結合：更新操作最佳實務

actor "客戶端" as client
participant "API Gateway" as gateway
participant "認證服務" as auth
participant "業務服務" as service
database "資料庫" as db
queue "訊息佇列" as mq

client -> gateway : HTTP 請求
gateway -> auth : 驗證 Token
auth --> gateway : 認證結果

alt 認證成功
    gateway -> service : 轉發請求
    service -> db : 查詢/更新資料
    db --> service : 回傳結果
    service -> mq : 發送事件
    service --> gateway : 回應資料
    gateway --> client : HTTP 200 OK
else 認證失敗
    gateway --> client : HTTP 401 Unauthorized
end

@enduml

這張流程圖展示了資料處理的流程，從輸入資料開始，經過驗證，最後儲存或顯示錯誤。

持續整合測試、程式碼風格檢查和型別檢查，並結合 Makefile 和 Git Hook，能有效提升開發效率和程式碼品質，為開發高品質的 Python 應用程式奠定堅實的基礎。