Python資料塑形與管理：Pydantic 與 SQLAlchemy 的完美結合：SQLAlchemy 與 Pydantic 的整合：開發強健的資料管理方案 (第6集)

SQLAlchemy 與 Pydantic 的整合：開發強健的資料管理方案

在上一篇文章中，我們探討了 SQLAlchemy 的 schema 功能。現在，我們將更進一步，引入 Pydantic 來強化我們的工具函式庫。這個強大的組合能讓我們為應用程式建立更強健、型別安全與易於維護的資料模型。我們將回顧 SQLAlchemy 的一些關鍵概念，介紹 Pydantic 的 schema 系統，並示範如何讓這兩個函式庫無縫協作。此外，我們還會探討使用 SQLModel 的優勢，它結合了兩者的精華。

SQLAlchemy Schemas 回顧與 Pydantic 簡介

在探討 Pydantic 與 SQLAlchemy 的整合之前，讓我們快速回顧 SQLAlchemy 中 schema 的概念，並介紹 Pydantic 中 schema 的概念。

在 SQLAlchemy 中，schema 指的是資料函式庫表的結構和組織。它定義了構成資料函式庫的表、欄位、關係和約束。精心設計的 schema 對於有效率的資料儲存、擷取和維護至關重要。

另一方面，Pydantic 引入了它自己的 schema 概念。在 Pydantic 中，schema 是一個根據類別的模型，它定義了資料物件的結構和驗證規則。SQLAlchemy schema 專注於資料函式庫結構，而 Pydantic schema 則用於資料驗證、序列化和反序列化。

SQLAlchemy 和 Pydantic schema 的主要區別在於它們的主要用途：

**SQLAlchemy schemas：**定義資料函式庫結構和 ORM 對應。
**Pydantic schemas：**定義具有驗證和序列化功能的資料模型。

讓我們回顧一個精心設計的 SQLAlchemy schema 的基本範例：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author = relationship("User", back_populates="posts")

內容解密：

這個 schema 定義了兩個表：users 和 posts，它們之間存在一對多關係。它透過以下方式展示了良好的 schema 設計：

使用適當的欄位型別和約束。
定義表之間的關係。
透過外部索引鍵約束確保資料完整性。

Pydantic 與 SQLAlchemy 的整合

現在，讓我們探討如何透過整合 Pydantic 來增強這個 schema。藉由結合這兩個函式庫，我們可以獲得以下好處：

型別安全和驗證： Pydantic 提供了強大的資料驗證功能。

  graph LR
    B[B]
    A[SQLAlchemy Schema] --> B{整合}
    C[Pydantic Schema] --> B
    B --> D[強健資料模型]

圖表說明: SQLAlchemy Schema 與 Pydantic Schema 整合後，能建立更強健的資料模型。

由於篇幅限制，我將在下一篇文章中繼續探討 Pydantic 與 SQLAlchemy 的整合細節，以及如何使用 SQLModel 創造更簡潔高效的資料模型。敬請期待！

深入 Python 領域：SQLAlchemy 與 Pydantic 的完美融合

在構建複雜的 Python 應用程式時，資料模型的管理至關重要。SQLAlchemy 提供了強大的物件關聯對映（ORM）功能，而 Pydantic 則以資料驗證和序列化見長。如何將兩者結合，開發更穩健、型別安全與易於維護的資料模式？本文將探討這個議題，並分享我多年來的實踐經驗。

SQLAlchemy 與 Pydantic 的優勢互補

SQLAlchemy 允許我們以物件導向的方式操作資料函式庫，簡化了資料函式庫互動的複雜性。Pydantic 則提供了資料驗證和序列化功能，確保資料的完整性和一致性。兩者結合，能帶來以下優勢：

序列化和反序列化： 輕鬆在 Python 物件和 JSON 之間轉換。
職責分離： 使用 SQLAlchemy 處理資料函式庫操作，Pydantic 負責資料驗證和 API 介面。
提升開發體驗： 更好的 IDE 支援和型別提示。

以下示例展示瞭如何整合 SQLAlchemy 和 Pydantic 模型：

from typing import List, Optional
from pydantic import BaseModel, EmailStr, constr
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# SQLAlchemy 模型
class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    posts = relationship("PostORM", back_populates="author")

class PostORM(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author = relationship("UserORM", back_populates="posts")

# Pydantic 模型
class PostBase(BaseModel):
    title: constr(min_length=1, max_length=100)
    content: str

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    author_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    username: constr(min_length=3, max_length=50)
    email: EmailStr

class UserCreate(UserBase):
    pass

class User(UserBase):
    id: int
    posts: List[Post] = []

    class Config:
        orm_mode = True

內容解密：

這段程式碼定義了 SQLAlchemy 的 ORM 模型 (UserORM 和 PostORM) 以及 Pydantic 模型 (User、UserCreate、Post 和 PostCreate)。這種分離允許我們使用 SQLAlchemy 模型進行資料函式庫操作，使用 Pydantic 模型進行輸入驗證和 API 回應，並利用 Pydantic 強大的驗證功能。orm_mode = True 則實作了 ORM 模型和 Pydantic 模型之間的輕鬆轉換。

SQLModel：兩全其美的方案

SQLModel 將 SQLAlchemy 和 Pydantic 的功能整合到單個套件中，讓我們可以定義同時作為 SQLAlchemy ORM 模型和 Pydantic 模型的模型。以下是用 SQLModel 重寫的示例：

from typing import List, Optional
from sqlmodel import Field, Relationship, SQLModel

class Post(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(max_length=100)
    content: str
    author_id: int = Field(foreign_key="user.id")
    author: "User" = Relationship(back_populates="posts")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(max_length=50)
    email: str = Field(max_length=120)
    posts: List[Post] = Relationship(back_populates="author")

# API 輸入/輸出模型
class PostCreate(SQLModel):
    title: str
    content: str

class UserCreate(SQLModel):
    username: str
    email: str

class PostRead(SQLModel):
    id: int
    title: str
    content: str
    author_id: int

class UserRead(SQLModel):
    id: int
    username: str
    email: str
    posts: List[PostRead] = []

內容解密：

使用 SQLModel 簡化了程式碼，同時保留了型別安全、驗證功能以及與 SQLAlchemy 和 Pydantic 生態系統的相容性。我們仍然可以建立單獨的輸入和輸出模型。

模組化管理 Schema 定義

為了提高專案的可維護性，建議將 Schema 定義儲存在名為 core 的子模組中，並為每個領域建立一個資料夾。以下是一種可能的專案結構：

  myproject/
│
├── core/
│ ├── __init__.py
│ ├── user/
│ │ ├── __init__.py
│ │ └── schema.py
│ └── post/
│ ├── __init__.py
│ └── schema.py
│
├── api/
│ ├── __init__.py
│ ├── user.py
│ └── post.py
│
└── main.py

這樣的組織結構可以更好地分離關注點、簡化複雜 Schema 的管理、提高程式碼可重用性，並使專案結構更清晰。

演進式 Schema 管理的挑戰

隨著專案的發展，Schema 的變更管理變得越來越重要。管理大型專案中的 Schema 變更面臨以下挑戰：

一致性： 確保應用程式的各個部分使用最新的 Schema 版本。
向後相容性： 在引入新功能的同時支援舊的 Schema 版本。
資料遷移： 將現有資料遷移以適應新的 Schema 結構。
檔案： 保持 Schema 檔案的更新。
測試： 驗證 Schema 變更不會破壞現有功能。
佈署： 將 Schema 更新與應用程式佈署協調。

除了資料函式庫結構之外，Schema 還包含 API 請求和回應、設定檔、訊息佇列、快取結構和資料序列化格式等。

在下一篇文章中，我將探討如何應對這些挑戰，並分享一些實用的 Schema 演進策略。

  graph LR
    B[B]
    A[SQLAlchemy] --> B{SQLModel};
    C[Pydantic] --> B;
    B --> D[簡潔程式碼];
    B --> E[型別安全];
    B --> F[資料驗證];

圖表說明: SQLModel 結合了 SQLAlchemy 和 Pydantic 的優點，帶來更簡潔的程式碼、型別安全和資料驗證功能。

  
在現代軟體開發中，資料的有效性和安全性至關重要。本文將探討如何利用 Pydantic 和 SQLAlchemy 構築兼具效能和安全性的應用程式。我將分享個人經驗和最佳實踐，涵蓋模式版本控制、資料遷移以及輸入驗證等關鍵環節。

## 模式版本控制的重要性

隨著專案的發展，資料模式的變更不可避免。妥善的版本控制能確保系統的穩定性和相容性。以下是一些實用的版本控制策略：

### 加入版本號

最簡單的方法是在模式定義中加入版本號：

```python
from pydantic import BaseModel

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

class UserV2(BaseModel):
    id: int
    first_name: str
    last_name: str

資料遷移指令碼

如同 Alembic 對資料函式庫的遷移，非資料函式庫模式也需要遷移指令碼。以下是一個自定義遷移系統的範例：

import json
from typing import Callable, Dict

migrations: Dict[str, Callable] = {}

def register_migration(from_version: str, to_version: str):
    def decorator(func):
        migrations[(from_version, to_version)] = func
        return func
    return decorator

@register_migration("1.0", "2.0")
def migrate_user_v1_to_v2(data: Dict) -> Dict:
    return {
        "id": data["id"],
        "first_name": data["name"].split()[0],
        "last_name": data["name"].split()[-1] if len(data["name"].split()) > 1 else ""
    }

def migrate_data(data: Dict, from_version: str, to_version: str) -> Dict:
    migration_key = (from_version, to_version)
    if migration_key in migrations:
        return migrations[migration_key](data)
    else:
        raise ValueError(f"No migration path from {from_version} to {to_version}")

# Example usage
old_data = {"id": 1, "name": "John Doe"}
new_data = migrate_data(old_data, "1.0", "2.0")
print(json.dumps(new_data, indent=2))

內容解密：這段程式碼定義了一個遷移系統，允許註冊不同版本間的遷移函式。register_migration 裝飾器將遷移函式與版本號關聯起來。migrate_data 函式根據版本號呼叫對應的遷移函式，實作資料的轉換。

模式登入檔

中央模式登入檔可以管理和版本化專案中的所有模式。

from typing import Dict, Type
from pydantic import BaseModel

class SchemaRegistry:
    _schemas: Dict[str, Dict[str, Type[BaseModel]]] = {}

    @classmethod
    def register(cls, name: str, version: str, schema: Type[BaseModel]):
        if name not in cls._schemas:
            cls._schemas[name] = {}
        cls._schemas[name][version] = schema

    @classmethod
    def get(cls, name: str, version: str) -> Type[BaseModel]:
        return cls._schemas[name][version]

# Register schemas
SchemaRegistry.register("User", "1.0", UserV1)
SchemaRegistry.register("User", "2.0", UserV2)

# Use schemas
user_schema_v1 = SchemaRegistry.get("User", "1.0")
user_schema_v2 = SchemaRegistry.get("User", "2.0")

內容解密：這段程式碼建立了一個 SchemaRegistry 類別，用於註冊和取得不同版本的 Pydantic 模型。透過 register 方法，可以將模型與名稱和版本號關聯起來。get 方法則可以根據名稱和版本號取得對應的模型。

輸入驗證：守護應用程式安全的第一道防線

輸入驗證是構建安全應用程式的關鍵。Pydantic 提供了強大的驗證工具，可以有效提升應用程式的安全性。

未經檢查的使用者輸入的風險

在與資料函式庫互動的複雜應用程式中，未經檢查的使用者輸入可能導致嚴重的安全漏洞，例如 SQL 注入和遠端程式碼執行。

追蹤使用者輸入的最佳實踐

為了降低風險並清楚追蹤使用者提供的資料，建議在所有包含使用者輸入的變數名稱後加上底線 _。

user_response_ = input("Enter your name: ")
processed_name = sanitize_input(user_response_)

透過以上技巧，可以有效管理模式演變，確保資料一致性，並提升應用程式的安全性。

  graph LR
    A[版本控制] --> B(加入版本號)
    A --> C(遷移指令碼)
    A --> D(模式登入檔)
    E[輸入驗證] --> F(風險評估)
    E --> G(最佳實踐)

在設計分散式系統時，我發現模式版本控制和輸入驗證是兩個容易被忽視但卻至關重要的環節。妥善的模式管理可以減少系統維護成本，而嚴格的輸入驗證則可以有效防範安全風險。

持續整合和交付（CI/CD）流程中，自動化的模式驗證和遷移測試是不可或缺的。這可以幫助團隊及早發現問題，確保系統的穩定性和可靠性。

  
在現今的軟體開發環境中，確保資料的有效性和安全性至關重要。Python 的 Pydantic 函式庫提供了一種優雅與高效能的方式來實作資料驗證和清洗，有效防止潛在的安全漏洞。本文將探討 Pydantic 的各種功能，並分享如何在實際專案中應用這些技術。

## 追蹤使用者輸入，強化程式碼安全性

我習慣在程式碼中使用一個特殊標記來追蹤使用者輸入，例如在變數名稱後面加上底線 `_`。這種方法有幾個好處：

1.  清晰標示使用者輸入，方便識別需要驗證的地方。
2.  追蹤使用者資料在應用程式中的流程，簡化除錯和安全稽核。
3.  確保所有使用者輸入在使用前都經過適當的清洗。

以下是一個簡單的範例：

```python
user_input_ = input("請輸入您的使用者名稱：")
# ... 後續處理 ...

經過驗證和清洗後，移除底線 _ 表示資料安全可使用：

safe_user_input = validate_and_sanitize(user_input_)

使用 Pydantic 模型進行輸入驗證

Pydantic 模型提供了一種結構化的方式來定義資料結構和驗證規則。以下是一個使用 Pydantic 進行輸入驗證的範例：

from pydantic import BaseModel, Field, validator
import re

class SafeUserInput(BaseModel):
    raw_input_: str = Field(..., min_length=1, max_length=100)
    sanitized_input: str = ""

    @validator('raw_input_')
    def check_for_sql_injection(cls, v):
        sql_keywords = ['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'DROP', 'UNION', 'FROM', 'WHERE']
        if any(keyword in v.upper() for keyword in sql_keywords):
            raise ValueError("偵測到潛在的 SQL 注入攻擊")
        return v

    @validator('raw_input_')
    def check_for_special_characters(cls, v):
        if re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError("不允許特殊字元")
        return v

    @validator('sanitized_input', always=True)
    def sanitize_input(cls, v, values):
        raw_input = values.get('raw_input_', '')
        sanitized = re.sub(r'[^a-zA-Z0-9\s]', '', raw_input)[:100]
        return sanitized.strip()

# 使用範例
user_input_ = "SELECT * FROM users; --"
try:
    safe_input = SafeUserInput(raw_input_=user_input_)
    print(f"已清洗的輸入：{safe_input.sanitized_input}")
except ValueError as e:
    print(f"無效的輸入：{e}")

內容解密：

這個 SafeUserInput 模型會檢查輸入中是否有潛在的 SQL 注入攻擊，並移除特殊字元。sanitize_input 函式會移除非字母數字字元，並限制輸入長度。請注意，這些函式僅為範例，並非安全或穩健的實作。

Pydantic 驗證器型別與應用場景

Pydantic 提供了多種型別的驗證器，適用於不同的場景：

欄位驗證器 (Field Validators): 驗證單個欄位，例如檢查字串長度或格式。
根驗證器 (Root Validators): 同時驗證多個欄位，例如檢查兩個欄位之間的關係。
前置根驗證器 (Pre-root Validators): 在其他驗證器之前執行，例如預處理輸入資料。
類別驗證器 (Class Validators): 應用於整個模型類別，例如將空字串轉換為 None。
總是驗證器 (Always Validators): 即使欄位未包含在輸入資料中也會執行，例如設定預設值。
使用 Annotated 欄位的驗證器: Pydantic 1.9+ 版本支援使用 Annotated 定義欄位約束和驗證器，使程式碼更簡潔易讀。

使用 Annotated 進行欄位驗證

Annotated 型別允許您將中繼資料附加到欄位的型別提示中，Pydantic 會使用這些中繼資料進行驗證。

from typing import Annotated
from pydantic import BaseModel, Field, StringConstraints, ValidationInfo, field_validator
import re

# ... (程式碼範例，詳見原文)

內容解密：

這個範例展示瞭如何使用 Annotated、StringConstraints、Field 和自定義驗證器函式來定義欄位驗證規則。

建立可重複使用的 Annotated 型別

您可以建立帶有註解的自定義型別，以便在多個模型中重複使用常見的驗證模式。

from typing import Annotated
from pydantic import StringConstraints, Field

Username = Annotated[str, StringConstraints(min_length=3, max_length=20), Field(description="User's login name")]
Email = Annotated[str, Field(pattern=r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')]

class User(BaseModel):
    username: Username
    email: Email

內容解密：

這個範例定義了 Username 和 Email 兩種可重複使用的 Annotated 型別，簡化了模型定義。

結合 Annotated 欄位與傳統驗證器

您可以將傳統驗證器方法與 Annotated 欄位結合使用，特別適用於涉及多個欄位或需要額外上下文資訊的複雜驗證。

from typing import Annotated
from pydantic import BaseModel, Field, field_validator

# ... (程式碼範例，詳見原文)

內容解密：

這個範例結合了 Annotated 欄位和傳統驗證器方法，實作了更全面的資料驗證。

透過 Pydantic 提供的豐富功能，您可以輕鬆地建立強大的資料驗證和清洗機制，提升應用程式的安全性，並減少潛在的錯誤。本文提供的技巧和最佳實踐，希望能幫助您更好地運用 Pydantic，開發更安全可靠的 Python 應用程式。

在現代軟體開發中，資料序列化是不可或缺的一環，尤其是在處理 Pydantic 和 SQLAlchemy 的整合時。這篇文章將探討模型序列化的必要性、內建功能的優缺點，以及如何利用進階技巧開發更強大的序列化方案。

序列化：資料流動的根本

模型序列化是將複雜資料結構或物件轉換成易於儲存、傳輸或重建格式的過程。在網頁應用程式和 API 的世界裡，序列化扮演著關鍵角色：

資料傳輸： 網路傳輸需要通用格式，序列化確保不同系統間的資料交換暢通無阻。
持久化： 將物件儲存到資料函式庫或檔案系統時，序列化提供必要的格式轉換。
互通性： 序列化讓不同程式語言和平台之間的資料分享成為可能。
快取： 序列化資料易於高效快取，加速資料讀取。
API 回應： 構建 API 時，序列化模型成 JSON 或其他格式是回應傳輸的標準做法。

內建序列化：便捷與限制

SQLAlchemy 和 Pydantic 都提供內建序列化功能，足以應付許多常見應用場景。

SQLAlchemy 的序列化利器

SQLAlchemy 提供以下幾種序列化模型例項的方法：

__dict__ 屬性：提供模型屬性的字典表示。

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

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

user = User(id=1, name="John Doe", email="john@example.com")
serialized_user = user.__dict__
print(serialized_user)
# 輸出：{'_sa_instance_state': <...>, 'id': 1, 'name': 'John Doe', 'email': 'john@example.com'}

# 內容解密：
# 這段程式碼展示瞭如何使用 SQLAlchemy 的 `__dict__` 屬性將 User 物件序列化成字典。
# 然而，輸出的字典包含了 SQLAlchemy 內部的狀態資訊 `_sa_instance_state`，這在實際應用中可能需要移除。

to_dict() 方法：您可以在模型上定義自訂的 to_dict() 方法。

class User(Base):
    # ... (同上)
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'email': self.email
        }

user = User(id=1, name="John Doe", email="john@example.com")
serialized_user = user.to_dict()
print(serialized_user)
# 輸出：{'id': 1, 'name': 'John Doe', 'email': 'john@example.com'}

# 內容解密：
# 這個例子展示瞭如何定義一個自訂的 `to_dict()` 方法來控制序列化的輸出。
# 這樣可以排除不需要的內部屬性，並只包含必要的欄位。

Pydantic 的序列化之道

Pydantic 模型也內建了序列化方法：

model_dump() 方法：將模型轉換為字典。

model_dump_json() 方法：將模型序列化為 JSON 字串。

from pydantic import BaseModel

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

user = UserModel(id=1, name="John Doe", email="john@example.com")
serialized_dict = user.model_dump()
serialized_json = user.model_dump_json()
print(serialized_dict)
# 輸出：{'id': 1, 'name': 'John Doe', 'email': 'john@example.com'}
print(serialized_json)
# 輸出：'{"id": 1, "name": "John Doe", "email": "john@example.com"}'


# 內容解密：
# Pydantic 的 `model_dump()` 和 `model_dump_json()` 方法提供了便捷的序列化方式，
# 可以輕鬆地將模型物件轉換成字典或 JSON 字串。

何時內建序列化遊刃有餘？

當以下條件成立時，內建序列化方法通常已足夠：

資料結構相對簡單。
不需要大幅客製化輸出格式。
使用易於序列化的標準資料型別。
不需要動態排除或包含特定欄位。
不處理迴圈參照或複雜關係。

內建序列化之不足

然而，在某些情況下，內建序列化方法力有未逮：

複雜關係：處理巢狀物件或多對多關係時，內建方法可能無法正確處理。
自訂資料型別：自訂資料型別可能無法直接序列化。
欄位條件式包含：可能需要根據特定條件包含或排除欄位。
欄位重新命名：有時需要在序列化輸出中更改欄位名稱。
計算欄位：可能需要包含即時計算而非儲存在資料函式庫中的欄位。
敏感資料：可能需要在序列化過程中排除或遮罩敏感資訊。

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。