使用者指南¶
深入了解 AutoCRUD 的功能,掌握從原型到生產的完整開發流程。
🎯 核心概念:自動 API 生成¶
AutoCRUD 的核心價值在於自動從資料模型生成完整的 REST API。這不僅僅是簡單的 CRUD 操作,而是包含驗證、文檔、分頁、查詢的 API 解決方案。
系統架構¶
AutoCRUD 提供兩個層次的介面:
AutoCRUD
:主要推薦!多模型 API 管理系統,自動生成 REST APISingleModelCRUD
:單一模型的程式化 CRUD 操作(當你需要細粒度控制時使用)
🚀 推薦開發流程¶
1. 從 AutoCRUD 開始¶
from autocrud import AutoCRUD
from dataclasses import dataclass
@dataclass
class User:
id: str
name: str
email: str
# 主要開發入口
crud = AutoCRUD()
crud.register_model(User) # 🎯 完整 REST API 自動生成
app = crud.create_fastapi_app(title="我的 API")
# 訪問 http://localhost:8000/docs 查看自動生成的 API
2. 當需要程式化控制時¶
# AutoCRUD 同時提供程式化介面
user_id = crud.create("users", {"name": "Alice", "email": "alice@example.com"})
user = crud.get("users", user_id)
# 如果需要更細粒度的控制,可以獲取單模型 CRUD
user_crud = crud.get_crud("users") # 返回 SingleModelCRUD 實例
🎨 資料模型支援¶
AutoCRUD 支援所有主流 Python 資料模型格式,讓你用喜歡的方式定義資料結構。重要:所有模型都必須包含 id
欄位。
🔥 推薦:使用 Pydantic 獲得進階驗證¶
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
from datetime import datetime
class User(BaseModel):
id: str
name: str = Field(..., min_length=1, max_length=100, description="使用者姓名")
email: EmailStr # 自動 email 格式驗證
age: Optional[int] = Field(None, ge=0, le=150, description="年齡")
is_active: bool = Field(True, description="是否啟用")
@validator('name')
def name_must_not_be_empty(cls, v):
if not v.strip():
raise ValueError('姓名不能為空')
return v.strip()
# 註冊後自動獲得完整的驗證 API
crud = AutoCRUD()
crud.register_model(User) # 🎯 所有驗證規則自動應用到 API
📦 簡潔選擇:Dataclass¶
from dataclasses import dataclass
from typing import Optional
@dataclass
class Product:
id: str
name: str
price: float
category: str
stock: int = 0
# 同樣能獲得完整的 REST API
crud.register_model(Product) # -> 完整的 /products API
🔧 靈活選擇:TypedDict¶
from typing import TypedDict, Optional
class Order(TypedDict):
id: str
user_id: str
items: list
total: float
status: str
crud.register_model(Order) # -> 完整的 /orders API
⚙️ 企業級功能配置¶
🕒 自動時間戳管理¶
讓 AutoCRUD 自動管理資料的創建和更新時間:
from autocrud import AutoCRUD, MetadataConfig
# 啟用自動時間戳
metadata_config = MetadataConfig(
enable_timestamps=True, # 自動添加 created_time 和 updated_time
enable_user_tracking=True, # 自動添加 created_by 和 updated_by(如果需要)
timestamp_field_names={
"created_time": "created_time",
"updated_time": "updated_time"
}
)
crud = AutoCRUD(metadata_config=metadata_config)
crud.register_model(User)
# 創建資料時自動添加時間戳
user_id = crud.create("users", {"name": "Alice", "email": "alice@example.com"})
user = crud.get("users", user_id)
print(user["created_time"]) # 2024-01-15T10:30:00Z
print(user["updated_time"]) # 2024-01-15T10:30:00Z
# 你的 API 也自動支援時間範圍查詢
# GET /users?created_time_start=2024-01-01&created_time_end=2024-12-31
🔍 高級查詢功能¶
AutoCRUD 自動為你的 API 提供豐富的查詢功能:
from autocrud import ListQueryParams, SortOrder
from datetime import datetime, timedelta
# 程式化查詢範例
params = ListQueryParams(
page=1,
page_size=20,
sort_by="created_time",
sort_order=SortOrder.DESC,
created_time_start=datetime.now() - timedelta(days=7), # 最近7天
created_time_end=datetime.now()
)
result = crud.list("users", params)
print(f"找到 {result.total} 個最近創建的使用者")
# 這些查詢功能自動對應到 REST API:
# GET /users?page=1&page_size=20&sort_by=created_time&sort_order=desc
# &created_time_start=2024-01-08T10:00:00Z
# &created_time_end=2024-01-15T10:00:00Z
🏢 生產環境:資料持久化¶
開發環境:記憶體存儲¶
適用於原型開發和測試:
# 預設使用記憶體存儲,重啟後資料消失
crud = AutoCRUD() # 開發時最簡單
crud.register_model(User)
生產環境:磁碟存儲¶
確保資料持久化,適用於生產環境:
from autocrud.storage import MemoryStorage
storage = MemoryStorage()
特點:
快速存取
程式結束後資料消失
適合測試和原型開發
硬碟儲存 (DiskStorage)¶
適用於持久化資料:
from autocrud.storage import DiskStorage
storage = DiskStorage(
storage_dir="./data",
serializer_type="json" # 或 "pickle", "msgpack"
)
特點:
資料持久化
支援多種序列化格式
自動建立儲存目錄
儲存工廠系統¶
AutoCRUD 支援儲存工廠來為不同資源創建專用的儲存後端:
from autocrud import DefaultStorageFactory
# 使用預設工廠(每個資源使用獨立的記憶體儲存)
factory = DefaultStorageFactory()
multi_crud = AutoCRUD(storage_factory=factory)
查詢和分頁¶
AutoCRUD 支援複雜的查詢、分頁和排序:
from autocrud import ListQueryParams, SortOrder
# 基本分頁查詢
query_params = ListQueryParams(
page=1,
page_size=20,
sort_by="created_time",
sort_order=SortOrder.DESC
)
result = user_crud.list(query_params)
print(f"總數: {result.total}")
print(f"當前頁: {result.page}")
print(f"項目: {result.items}")
# 時間範圍查詢
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)
query_params = ListQueryParams(
created_time_start=now - timedelta(days=7),
created_time_end=now,
sort_by="name"
)
recent_users = user_crud.list(query_params)
序列化格式¶
JSON (推薦)¶
storage = DiskStorage(serializer_type="json")
人類可讀
跨語言相容
較小的檔案大小
Pickle¶
storage = DiskStorage(serializer_type="pickle")
Python 原生支援
支援任意 Python 對象
僅限 Python 使用
MessagePack¶
storage = DiskStorage(serializer_type="msgpack")
二進位格式
高效壓縮
跨語言支援
多模型管理¶
基本用法¶
from autocrud import AutoCRUD, ResourceNameStyle
from autocrud.storage import MemoryStorage
# 建立多模型系統
multi_crud = AutoCRUD(
resource_name_style=ResourceNameStyle.SNAKE, # 命名風格
use_plural=True # 預設使用複數形式
)
# 註冊多個模型
user_crud = multi_crud.register_model(User) # 返回 SingleModelCRUD 實例
product_crud = multi_crud.register_model(Product)
order_crud = multi_crud.register_model(Order)
資源名稱自訂¶
# 自動產生複數形式 (預設)
multi_crud.register_model(User) # -> users
# 指定單數形式
multi_crud.register_model(Product, use_plural=False) # -> product
# 完全自訂名稱
multi_crud.register_model(Company, resource_name="organizations") # -> organizations
# 不同的命名風格
multi_crud = AutoCRUD(resource_name_style=ResourceNameStyle.CAMEL)
multi_crud.register_model(UserProfile) # -> userProfiles
跨模型操作¶
# 直接在多模型系統上執行操作
user_id = multi_crud.create("users", {"name": "Alice", "email": "alice@example.com"})
product_id = multi_crud.create("products", {"name": "Laptop", "price": 999.99})
# 取得特定模型的 CRUD 實例
user_crud = multi_crud.get_resource_crud("users")
all_users = user_crud.list_all()
# 列出所有註冊的資源
resources = multi_crud.list_resources()
插件系統¶
AutoCRUD 支援可擴展的插件系統:
使用預設插件¶
from autocrud import DEFAULT_PLUGINS, plugin_manager
# 預設插件包含:create, get, update, delete, count, list
print([plugin.name for plugin in DEFAULT_PLUGINS])
# 確保註冊預設插件
from autocrud import ensure_default_plugins_registered
ensure_default_plugins_registered()
創建自訂插件¶
from autocrud import BaseRoutePlugin, PluginRouteConfig, RouteMethod
from fastapi import BackgroundTasks
class HealthCheckPlugin(BaseRoutePlugin):
def __init__(self):
super().__init__("health", "1.0.0")
def get_routes(self, crud):
async def health_handler(crud, background_tasks: BackgroundTasks):
return {"status": "healthy", "resource": crud.resource_name}
return [PluginRouteConfig(
name="health",
path="/health",
method=RouteMethod.GET,
handler=health_handler,
summary="健康檢查",
priority=1 # 高優先級
)]
# 註冊插件
from autocrud import plugin_manager
plugin_manager.register_plugin(HealthCheckPlugin())
高級更新系統¶
AutoCRUD 支援原子操作的高級更新:
from autocrud import AdvancedUpdater, set_value, list_add, dict_update
updater = AdvancedUpdater()
# 定義更新操作
operations = [
set_value("status", "active"),
list_add("tags", "premium"),
dict_update("metadata", {"last_login": "2024-01-01"})
]
# 執行原子更新
success = updater.update(user_crud, user_id, operations)
FastAPI 整合¶
單模型 API¶
from autocrud import SingleModelCRUD
from autocrud.fastapi_generator import FastAPIGenerator
user_crud = SingleModelCRUD(model=User, storage=storage, resource_name="users")
# 使用 FastAPIGenerator
generator = FastAPIGenerator()
app = generator.create_app(
crud_systems=[user_crud],
title="User API",
description="使用者管理 API",
version="1.0.0"
)
多模型 API¶
from autocrud import AutoCRUD
multi_crud = AutoCRUD()
multi_crud.register_model(User)
multi_crud.register_model(Product)
app = multi_crud.create_fastapi_app(
title="多模型 API",
description="統一的 CRUD API",
prefix="/api/v1"
)
路由配置¶
可以透過 RouteConfig 自訂路由行為:
from autocrud import RouteConfig, RouteOptions
# 自訂路由配置
route_config = RouteConfig(
create=RouteOptions.enabled_route(),
get=RouteOptions.enabled_route(),
update=RouteOptions.disabled_route(), # 禁用更新
delete=RouteOptions.disabled_route(), # 禁用刪除
list=RouteOptions.enabled_route(),
count=RouteOptions.enabled_route()
)
user_crud = SingleModelCRUD(
model=User,
storage=storage,
resource_name="users",
route_config=route_config
)
自訂路由前綴¶
app = multi_crud.create_fastapi_app(
prefix="/api/v2" # 所有路由將以 /api/v2 開頭
)
錯誤處理¶
AutoCRUD 提供了統一的錯誤處理機制:
# CRUD 方法的返回值
user_id = user_crud.create(data) # 成功返回 ID,失敗拋出異常
user = user_crud.get(user_id) # 成功返回資料,不存在返回 None
success = user_crud.update(user_id, data) # 成功返回 True,失敗返回 False
success = user_crud.delete(user_id) # 成功返回 True,失敗返回 False
# FastAPI 路由會自動處理異常並返回適當的HTTP狀態碼
效能最佳化¶
大量資料處理¶
# 使用 list_all 時要小心記憶體使用
all_users = user_crud.list_all() # 可能消耗大量記憶體
# 建議使用分頁查詢
query_params = ListQueryParams(page=1, page_size=100)
page_result = user_crud.list(query_params)
儲存後端選擇¶
# 開發和測試:使用記憶體儲存
storage = MemoryStorage()
# 生產環境:使用硬碟儲存搭配 JSON 序列化
storage = DiskStorage(storage_dir="./data", serializer_type="json")
# 高效能需求:使用 MessagePack 序列化
storage = DiskStorage(storage_dir="./data", serializer_type="msgpack")
ID 產生器¶
預設 UUID 產生器¶
# 使用預設的 UUID4 產生器
crud = SingleModelCRUD(model=User, storage=storage, resource_name="users")
自訂 ID 產生器¶
def custom_id_generator():
import time
return f"user_{int(time.time())}"
crud = SingleModelCRUD(
model=User,
storage=storage,
resource_name="users",
id_generator=custom_id_generator
)
序列 ID 產生器¶
def sequential_id_generator():
if not hasattr(sequential_id_generator, 'counter'):
sequential_id_generator.counter = 0
sequential_id_generator.counter += 1
return str(sequential_id_generator.counter)
crud = SingleModelCRUD(
model=User,
storage=storage,
resource_name="users",
id_generator=sequential_id_generator
)
錯誤處理¶
常見異常¶
from autocrud.exceptions import AutoCRUDError, ValidationError, StorageError
try:
user = crud.create(invalid_data)
except ValidationError as e:
print(f"資料驗證錯誤: {e}")
except StorageError as e:
print(f"儲存錯誤: {e}")
except AutoCRUDError as e:
print(f"通用錯誤: {e}")
資料驗證¶
# Pydantic 模型會自動進行資料驗證
from pydantic import BaseModel, validator
class User(BaseModel):
name: str
age: int
@validator('age')
def validate_age(cls, v):
if v < 0:
raise ValueError('年齡不能為負數')
return v
最佳實踐¶
1. 模型設計¶
from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
@dataclass
class User:
name: str
email: str
age: Optional[int] = None
is_active: bool = True
created_at: datetime = field(default_factory=datetime.now)
tags: List[str] = field(default_factory=list)
2. 儲存選擇¶
開發/測試: 使用
MemoryStorage
生產環境: 使用
DiskStorage
配合適當的序列化格式大量資料: 考慮實作自訂儲存後端
3. API 設計¶
# 為不同的資源使用合適的 URL 形式
multi_crud.register_model(User) # RESTful: /users
multi_crud.register_model(Config, use_plural=False) # Singleton: /config
multi_crud.register_model(Company, resource_name="organizations") # Custom: /organizations
4. 錯誤處理¶
from fastapi import HTTPException
def safe_create_user(data: dict):
try:
return user_crud.create(data)
except ValidationError as e:
raise HTTPException(status_code=400, detail=str(e))
except StorageError as e:
raise HTTPException(status_code=500, detail="儲存錯誤")
效能考慮¶
記憶體使用¶
MemoryStorage
將所有資料保存在記憶體中對於大量資料,考慮使用
DiskStorage
序列化效能¶
JSON: 平衡效能和可讀性
Pickle: 最快,但僅限 Python
MessagePack: 高效的二進位格式
併發存取¶
目前的實作不是執行緒安全的。在高併發環境中:
使用適當的鎖機制
考慮使用資料庫後端
實作自訂的執行緒安全儲存後端