進階序列化：突破限制
當內建序列化不足時，可以採用以下策略強化序列化過程：
- 自訂序列化方法：建立處理複雜情況的自訂序列化方法。
- 序列化 Mixin：建立提供序列化功能的 Mixin 類別。
- 使用 Pydantic 模型定義 SQLAlchemy 模型的序列化 schema。
- 利用 Pydantic 的進階功能，例如欄位重新命名、排除、自訂編碼器、計算欄位和條件式欄位包含。
這篇文章探討了模型序列化在 Pydantic 和 SQLAlchemy 整合中的重要性，並介紹了內建方法和進階技巧。根據專案需求選擇合適的序列化策略，才能有效管理資料並提升應用程式效能。
## 資料治理：建構有效資料管理的框架
在本章中，玄貓將探討穩健資料治理計畫的核心組成部分。我們將探討如何建立資料所有權、定義品質標準、實施存取控制、確保法規遵循性、建立生命週期管理策略以及開發治理指標。讀完本章後，您將全面瞭解如何實施和維護有效的資料治理框架。
### 建立資料所有權和管理權
資料所有權和管理權是有效資料治理的基礎。它們確保資料品質、安全性、以及正確使用方面有明確的責任歸屬。讓我們深入研究這些概念：
**資料所有者：** 資料所有者對資料的完整生命週期負有最終責任。他們通常是高階管理人員，負責定義資料的使用方式、存取方式以及保護方式。
**資料管理者：** 資料管理者負責資料的日常管理和維護。他們確保資料的品質、準確性和可用性。
**權責區分：** 資料所有者設定策略，而資料管理者執行這些策略。這種區分確保了資料治理的有效性和一致性。
### 定義資料品質標準
資料品質對於任何資料治理計畫都至關重要。高品質的資料是準確、完整、一致、及時與相關的。以下是如何定義資料品質標準：
**建立資料品質指標：** 這些指標應可衡量資料的準確性、完整性、一致性、及時性和相關性。
**實施資料品品檢查：** 這些檢查應定期執行，以識別和糾正資料品質問題。
**資料品質稽核：** 稽核應定期進行，以評估資料品質標準的有效性。
### 實施資料存取控制
資料存取控制對於保護敏感資料至關重要。有效的存取控制可確保只有授權人員才能存取資料。以下是如何實施資料存取控制：
**根據角色的存取控制（RBAC）：** RBAC 根據使用者的角色授予存取許可權。
**資料加密：** 加密可保護資料免受未經授權的存取。
**存取日誌：** 記錄所有資料存取嘗試，以識別任何可疑活動。
### 確保法規遵循性
資料治理計畫必須符合所有適用的法規。這包括資料隱私法規，例如 GDPR 和 CCPA。以下是如何確保法規遵循性：
**資料保護影響評估（DPIA）：** DPIA 識別和減輕資料隱私風險。
**資料洩露通知：** 如果發生資料洩露，必須立即通知相關機構。
**法規更新：** 持續關注資料隱私法規的變化，並更新您的資料治理計畫以符合這些變化。
### 建立資料生命週期管理策略
資料生命週期管理策略概述了資料在其整個生命週期中的管理方式，從建立到刪除。以下是如何建立資料生命週期管理策略：
**資料保留策略：** 定義資料的保留期限。
**資料歸檔策略：** 定義如何歸檔資料。
**資料刪除策略：** 定義如何刪除資料。
### 開發資料治理指標
資料治理指標用於衡量資料治理計畫的有效性。以下是如何開發資料治理指標：
**資料品質指標：** 衡量資料的品質。
**遵循性指標：** 衡量法規遵循性。
**安全性指標：** 衡量資料安全性。
```mermaid
graph LR
C[C]
A[資料所有者] --> B(設定策略)
B --> C{資料管理者}
C --> D[執行策略]
內容解密： 上面的 Mermaid 圖表展示了資料所有者和資料管理者之間的關係。資料所有者負責設定策略，而資料管理者負責執行這些策略。
總結：有效的資料治理對於任何組織都至關重要。透過遵循本章中概述的步驟，您可以實施和維護一個穩健的資料治理框架，以保護您的資料並確保其正確使用。
在數位時代，資料已成為企業最重要的資產之一。有效管理和運用資料，不僅能提升決策效率、驅動業務成長，更能降低風險、確保合規。本文將探討資料治理的各個導向，並提供 Python 程式碼範例，協助您建立完善的資料治理策略。
資料所有權與資料管理策略
資料所有者通常是高階主管，負責制定資料使用策略、核准資料存取請求、確保符合法規，並分配資料管理資源。以下程式碼範例展示如何定義資料資產及其所有者：
class DataAsset:
def __init__(self, name, owner, description):
self.name = name
self.owner = owner
self.description = description
customer_data = DataAsset(
name="客戶資料函式庫",
owner="資料關係副執行長 王小明",
description="包含所有客戶資訊，包括個人詳細資料和購買紀錄"
)
資料管理員則負責資料資產的日常管理，包括維護資料品質、實施資料治理策略、處理資料相關問題，以及提供中繼資料和檔案。以下程式碼範例展示資料管理員如何執行資料品品檢查：
from datetime import datetime
class DataQualityCheck:
def __init__(self, asset, steward):
self.asset = asset
self.steward = steward
self.timestamp = datetime.now()
self.issues = []
def add_issue(self, issue):
self.issues.append(issue)
def generate_report(self):
return (
f"資料品質報告 - {self.asset.name}\n"
f"管理員：{self.steward}\n"
f"日期：{self.timestamp}\n"
f"發現問題數：{len(self.issues)}\n"
f"詳細資訊：{', '.join(self.issues)}"
)
check = DataQualityCheck(customer_data, "資料分析師 李大華")
check.add_issue("5% 的電子郵件地址無效")
print(check.generate_report())
資料品質標準與驗證機制
資料品質標準確保資料符合預期用途。以下列出幾個關鍵導向，並以 Python 程式碼示範如何實作資料品品檢查：
- 準確性: 資料正確反映真實世界的實體或事件。
- 完整性: 所有必要的資料都存在。
- 一致性: 不同系統中的資料一致。
- 時效性: 資料保持最新狀態。
- 有效性: 資料符合定義的格式和型別。
from pydantic import BaseModel, EmailStr, validator
from typing import Optional
class CustomerData(BaseModel):
id: int
name: str
email: EmailStr
age: Optional[int]
@validator('name')
def name_must_be_complete(cls, v):
if len(v.split()) < 2:
raise ValueError('請提供完整姓名')
return v
@validator('age')
def age_must_be_reasonable(cls, v):
if v is not None and (v < 0 or v > 120):
raise ValueError('年齡必須介於 0 到 120 之間')
return v
# 使用範例
try:
customer = CustomerData(id=1, name="王小明", email="xiaoming@example.com", age=30)
print("有效的客戶資料:", customer)
except ValueError as e:
print("資料品質問題:", str(e))
資料存取控制與安全機制
資料存取控制確保只有授權人員才能檢視或修改資料。這包含驗證使用者身份、決定使用者可執行的操作，以及追蹤所有資料存取和修改。以下程式碼示範一個簡單的存取控制系統：
from enum import Enum
class AccessLevel(Enum):
READ = 1
WRITE = 2
ADMIN = 3
class User:
def __init__(self, name, access_level):
self.name = name
self.access_level = access_level
class DataAccessManager:
def __init__(self):
self.users = {}
def add_user(self, user):
self.users[user.name] = user
def can_access(self, user_name, required_level):
if user_name not in self.users:
return False
return self.users[user_name].access_level.value >= required_level.value
# 使用範例
dam = DataAccessManager()
dam.add_user(User("Alice", AccessLevel.ADMIN))
dam.add_user(User("Bob", AccessLevel.READ))
print("Alice 可以寫入嗎？", dam.can_access("Alice", AccessLevel.WRITE)) # True
print("Bob 可以寫入嗎？", dam.can_access("Bob", AccessLevel.WRITE)) # False
法規遵循與資料保護
法規遵循是指遵守資料管理相關的法律和法規，例如 GDPR、CCPA 和 HIPAA。以下程式碼範例展示一個同意管理系統：
from datetime import datetime, timedelta
class ConsentManager:
def __init__(self):
self.consents = {}
def give_consent(self, user_id, purpose, duration_days):
expiry = datetime.now() + timedelta(days=duration_days)
self.consents[(user_id, purpose)] = expiry
def check_consent(self, user_id, purpose):
key = (user_id, purpose)
if key not in self.consents:
return False
return datetime.now() < self.consents[key]
def revoke_consent(self, user_id, purpose):
key = (user_id, purpose)
if key in self.consents:
del self.consents[key]
# 使用範例
cm = ConsentManager()
cm.give_consent("user123", "marketing_emails", 30)
print("使用者同意行銷郵件嗎？", cm.check_consent("user123", "marketing_emails")) # True
cm.revoke_consent("user123", "marketing_emails")
print("使用者同意行銷郵件嗎？", cm.check_consent("user123", "marketing_emails")) # False
資料生命週期管理
資料生命週期管理涵蓋資料從建立到刪除的整個過程。以下程式碼範例展示資料生命週期策略的實作：
from enum import Enum
from datetime import datetime, timedelta
class DataStage(Enum):
ACTIVE = 1
ARCHIVED = 2
DELETED = 3
class DataRecord:
def __init__(self, data, creation_date):
self.data = data
self.creation_date = creation_date
self.stage = DataStage.ACTIVE
class DataLifecycleManager:
def __init__(self, active_period_days, archive_period_days):
self.active_period = timedelta(days=active_period_days)
self.archive_period = timedelta(days=archive_period_days)
self.records = []
def add_record(self, record):
self.records.append(record)
def update_lifecycle_stages(self):
now = datetime.now()
for record in self.records:
age = now - record.creation_date
if age > self.active_period + self.archive_period:
record.stage = DataStage.DELETED
elif age > self.active_period:
record.stage = DataStage.ARCHIVED
# 使用範例
dlm = DataLifecycleManager(active_period_days=30, archive_period_days=60)
dlm.add_record(DataRecord("客戶 A 資料", datetime.now() - timedelta(days=20)))
dlm.add_record(DataRecord("客戶 B 資料", datetime.now() - timedelta(days=50)))
dlm.add_record(DataRecord("客戶 C 資料", datetime.now() - timedelta(days=100)))
dlm.update_lifecycle_stages()
for record in dlm.records:
print(f"紀錄: {record.data}, 階段: {record.stage}")
資料治理指標和關鍵績效指標(KPI)
為了確保資料治理計畫的有效性，必須開發和追蹤相關指標和 KPI。以下程式碼範例展示一個簡單的資料治理儀錶板：
class DataGovernanceDashboard:
def __init__(self):
self.metrics = {
"data_quality_score": 0,
"policy_compliance_rate": 0,
"avg_issue_resolution_time": 0,
"avg_access_request_time": 0,
"data_catalog_completeness": 0
}
def update_metric(self, metric_name, value):
if metric_name in self.metrics:
self.metrics[metric_name] = value
def generate_report(self):
report = "資料治理儀錶板\n"
report += "==========================\n"
for metric, value in self.metrics.items():
report += f"{metric.replace('_', ' ').title()}: {value}\n"
return report
# 使用範例
dashboard = DataGovernanceDashboard()
dashboard.update_metric("data_quality_score", 85)
dashboard.update_metric("policy_compliance_rate", 92)
dashboard.update_metric("avg_issue_resolution_time", 48) # 小時
dashboard.update_metric("avg_access_request_time", 24) # 小時
dashboard.update_metric("data_catalog_completeness", 78)
print(dashboard.generate_report())
透過建立清晰的資料所有權和管理機制、定義品質標準、實施存取控制、確保法規遵循、管理資料生命週期以及追蹤相關指標，企業可以建立一個高效、安全與合規的資料生態系統，最大限度地發揮資料價值，同時將風險降至最低。
在現代 Web 應用程式開發中，RESTful API 扮演著至關重要的角色，而資料函式庫操作更是 API 的核心功能之一。本文將探討如何在 RESTful API 中使用 SQLAlchemy 和 Pydantic 進行資料的新增和讀取操作，涵蓋了從簡單的單表操作到複雜的多表關聯，並提供最佳實踐和效能最佳化建議。同時，也示範瞭如何結合 FastAPI 和 SQLModel 進行更簡潔高效的開發。
## 新增資料：從簡單到複雜
### 簡單案例：新增單一使用者
首先，我們從一個簡單的例子開始：新增使用者。假設我們已經定義了 `User` 模型和 `UserCreate` schema：
```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)
# core/user/schemas.py
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
username: str
email: EmailStr
API 端點程式碼如下：
# api/v1/endpoints/user.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from core.user import models, schemas
from db.session import get_db
router = APIRouter()
@router.post("/users/", response_model=schemas.UserCreate)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = models.User(username=user.username, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
內容解密：
這段程式碼定義了一個 POST 端點 /users/，接收 UserCreate schema 格式的資料，並將其新增到資料函式庫中。Depends(get_db) 注入資料函式庫 session，db.refresh(db_user) 確保傳回的物件包含新生成的 ID。
複雜案例：新增關聯資料
接下來，我們來看一個更複雜的例子：新增訂單及其包含的商品。
# 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")
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")
# core/order/schemas.py
from pydantic import BaseModel
from typing import List
class OrderItemCreate(BaseModel):
product_id: int
quantity: int
class OrderCreate(BaseModel):
user_id: int
items: List[OrderItemCreate]
API 端點程式碼如下：
# api/v1/endpoints/order.py
@router.post("/orders/", response_model=schemas.OrderCreate)
def create_order(order: schemas.OrderCreate, db: Session = Depends(get_db)):
db_order = models.Order(user_id=order.user_id)
db.add(db_order)
db.flush() # 先 flush 以取得 order ID
for item in order.items:
db_item = models.OrderItem(order_id=db_order.id, **item.dict())
db.add(db_item)
db.commit()
db.refresh(db_order)
return db_order
內容解密：
這段程式碼示範瞭如何處理多表關聯新增。db.flush() 在提交事務前先將 Order 物件寫入資料函式庫，以便取得 order_id，然後再新增 OrderItem 物件。
讀取資料：高效查詢
讀取資料是 API 的另一項核心功能。SQLAlchemy 提供了強大的查詢功能，可以滿足各種複雜的查詢需求。
(程式碼範例與內容解密將在後續章節中詳細介紹)
透過以上案例，我們展示瞭如何在 RESTful API 中使用 SQLAlchemy 和 Pydantic 進行資料的新增和讀取操作。後續章節將探討更進階的資料操作技巧，包括更新、刪除、分頁、排序、過濾等，並結合 FastAPI 和 SQLModel 框架，開發更強健、高效的 API。
在 RESTful API 的開發過程中，資料函式庫的讀取和更新操作至關重要。本文將探討如何利用 SQLAlchemy 和 Pydantic，以及 FastAPI 和 SQLModel，有效地執行這些操作。
讀取資料：從簡單到複雜
首先，我們以一個簡單的使用者讀取範例開始，使用先前定義的 User 模型和 Schema：
# 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 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()
@router.get("/users/{user_id}", response_model=schemas.UserRead)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
內容解密：
這段程式碼定義了使用者模型、Schema 和 API 端點。當收到 GET 請求時，它會根據提供的 user_id 查詢資料函式庫。如果找到使用者，則傳回使用者資訊；否則，傳回 404 錯誤。orm_mode = True 讓 Pydantic 自動將 SQLAlchemy 模型轉換為 Pydantic Schema。
接著，我們來看一個更複雜的例子，涉及多個關聯表格：讀取訂單及其相關專案。
# core/order/models.py
# ... (Order and OrderItem models)
# core/order/schemas.py
# ... (OrderRead and OrderItemRead schemas)
# api/v1/endpoints/order.py
@router.get("/orders/{order_id}", response_model=schemas.OrderRead)
def read_order(order_id: int, db: Session = Depends(get_db)):
db_order = db.query(models.Order).options(selectinload(models.Order.items)).filter(models.Order.id == order_id).first()
if db_order is None:
raise HTTPException(status_code=404, detail="Order not found")
return db_order
內容解密：
這個例子使用了 selectinload 來一次性載入關聯的專案，避免了 N+1 問題，提升了查詢效率。
讀取操作的最佳實務
在實作讀取操作時，以下是一些最佳實務：
- 使用查詢引數進行過濾和分頁： 提高查詢的靈活性。
- 在服務層實作資料函式庫操作： 分離關注點，使程式碼更具模組化。
- 使用聯結載入相關資料： 減少資料函式庫查詢次數，提高效能。
更新資料
以下是如何使用 SQLAlchemy 和 Pydantic 更新資料的示例：
# ... (schemas and models)
@router.put("/users/{user_id}", response_model=schemas.UserRead)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
# 更新使用者資料
for key, value in user.dict(exclude_unset=True).items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
內容解密：
此程式碼接收 PUT 請求，並根據提供的 user_id 更新使用者資訊。exclude_unset=True 確保只更新提供的欄位。
函式式與物件導向的程式設計方法
除了上述示例，我們還可以採用函式式或物件導向的程式設計方法來實作讀取和更新操作，搭配 SQLAlchemy/Pydantic 或 FastAPI/SQLModel。物件導向方法在需要維護狀態或封裝複雜查詢邏輯時特別有用。
系統架構
graph LR
API[API]
Client[Client]
Database[Database]
Service[Service]
API --> Service
Service --> Database
Client --> API
圖表說明： 此圖表展示了 API、服務層和資料函式庫之間的互動關係，以及客戶端如何與 API 進行互動。
透過以上示例和最佳實務，我們可以更有效地在 RESTful API 中使用 SQLAlchemy 和 Pydantic 進行資料函式庫的讀取和更新操作，並根據專案需求選擇合適的程式設計方法。
本文涵蓋了 RESTful API 中資料函式庫讀取和更新操作的關鍵導向，包括簡單和複雜資料關係的處理、最佳實務、函式式和物件導向方法，以及 FastAPI 和 SQLModel 的應用。希望這些資訊能幫助你更好地設計和開發高效能的 API。
在現代 Web 應用中，資料更新是不可或缺的功能。本文將以 FastAPI 為框架，結合 SQLAlchemy 和 SQLModel，深入解析資料更新的各種技巧，並提供最佳實務與效能最佳化策略。
更新操作：由淺入深
首先，我們從一個簡單的使用者更新案例開始，逐步深入到更複雜的多表關聯更新。
簡單案例：更新單一資料
以下範例示範如何更新系統中的使用者資料，使用的是先前定義的 User 模型和 Schema。
# 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 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()
@router.put("/users/{user_id}", response_model=schemas.UserRead)
def update_user(user_id: int, user_update: schemas.UserUpdate, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
update_data = user_update.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
內容解密：
這段程式碼定義了更新使用者資料的 API 端點。它接收使用者 ID 和更新資料，透過 SQLAlchemy 查詢資料函式庫，若使用者存在則更新對應欄位，最後提交變更並回傳更新後的使用者資料。exclude_unset=True 確保只更新提交的欄位，未提交的欄位保持不變。
複雜案例：更新關聯資料
接下來，我們探討更複雜的場景：更新訂單及其包含的商品。
# core/order/models.py
# ... (Order and OrderItem model definitions)
# core/order/schemas.py
# ... (OrderItemUpdate and OrderUpdate schema definitions)
# api/v1/endpoints/order.py
@router.put("/orders/{order_id}", response_model=schemas.OrderRead)
def update_order(order_id: int, order_update: schemas.OrderUpdate, db: Session = Depends(get_db)):
# ... (Order update logic as described in the original text)
內容解密：
此程式碼片段處理訂單及其商品的更新邏輯。它首先檢查訂單是否存在，然後更新訂單的主要資訊。對於商品，它會建立現有商品的字典，以便快速查詢。接著，它迭代提供的商品更新，若商品存在則更新其欄位，若為新商品則建立並新增到資料函式庫。最後，提交所有變更並重新整理訂單物件。
