SQLAlchemy ORM 原理深度解析:從 Unit of Work 到 Identity Map
0. 前言與生成說明
本文深入解析 Python 生態系統中最強大的 ORM 框架 —— SQLAlchemy 的內部實現原理。
文章目標
讀完本文,你將理解:
- SQLAlchemy 的整體架構與設計理念
- Session 與 Unit of Work 模式的實現
- Identity Map 如何保證對象唯一性
- Mapper 如何建立類與表的映射
- 屬性追蹤與變更檢測機制
- 關係載入策略的實現
1. SQLAlchemy 整體架構
SQLAlchemy 分為兩個主要層次:
| 層次 | 描述 | 核心組件 |
|---|---|---|
| Core | SQL 表達式語言與資料庫抽象 | Engine、Connection、SQL Expression |
| ORM | 物件關係映射層 | Session、Mapper、Query |
1.1 源代碼結構
sqlalchemy/lib/sqlalchemy/
├── __init__.py
├── engine/ # 資料庫引擎與連接池
│ ├── base.py
│ ├── create.py
│ └── result.py
├── sql/ # SQL 表達式語言
│ ├── elements.py
│ ├── selectable.py
│ └── expression.py
├── orm/ # ORM 核心
│ ├── session.py # Session 實現
│ ├── mapper.py # Mapper 類與表映射
│ ├── attributes.py # 屬性追蹤系統
│ ├── state.py # InstanceState 對象狀態
│ ├── identity.py # Identity Map
│ ├── unitofwork.py # Unit of Work
│ ├── loading.py # 對象載入
│ ├── relationships.py # 關係處理
│ ├── strategies.py # 載入策略
│ └── query.py # Query 對象
├── dialects/ # 資料庫方言
│ ├── postgresql/
│ ├── mysql/
│ └── sqlite/
└── pool/ # 連接池
1.2 架構概覽
flowchart TB
subgraph Application["應用層"]
APP["Python 應用"]
end
subgraph ORM["ORM 層"]
SESSION["Session<br/>Unit of Work"]
MAPPER["Mapper<br/>類-表映射"]
QUERY["Query<br/>查詢構建"]
IDENTITY["Identity Map<br/>對象緩存"]
STATE["InstanceState<br/>對象狀態追蹤"]
end
subgraph Core["Core 層"]
ENGINE["Engine<br/>資料庫引擎"]
CONN["Connection<br/>資料庫連接"]
SQL["SQL Expression<br/>SQL 表達式"]
POOL["Connection Pool<br/>連接池"]
end
subgraph DB["資料庫"]
DATABASE[("PostgreSQL / MySQL / SQLite")]
end
APP --> SESSION
APP --> QUERY
SESSION --> MAPPER
SESSION --> IDENTITY
SESSION --> STATE
QUERY --> SQL
MAPPER --> STATE
SESSION --> ENGINE
ENGINE --> CONN
ENGINE --> POOL
CONN --> DATABASE
SQL --> CONN
2. 核心設計模式
2.1 Data Mapper 模式
SQLAlchemy ORM 採用 Data Mapper 模式,將資料庫表與 Python 類完全解耦:
flowchart LR
subgraph Domain["領域對象"]
USER["User 類<br/>純 Python 對象"]
end
subgraph Mapper["Mapper 層"]
MP["Mapper<br/>映射配置"]
end
subgraph Database["資料庫"]
TB["users 表"]
end
USER <-->|"映射"| MP
MP <-->|"持久化"| TB
與 Active Record 模式對比:
| 模式 | 特點 | 框架範例 |
|---|---|---|
| Data Mapper | 領域對象不依賴 ORM | SQLAlchemy |
| Active Record | 領域對象繼承 ORM 基類 | Django ORM、Rails |
2.2 Unit of Work 模式
Unit of Work 是 SQLAlchemy 的核心模式,負責:
- 追蹤所有對象變更
- 計算變更順序(拓撲排序)
- 批量執行 SQL 語句
sequenceDiagram
participant App as 應用程式
participant Session as Session
participant UoW as Unit of Work
participant DB as 資料庫
App->>Session: session.add(user)
Session->>Session: 追蹤 user 對象
App->>Session: user.name = "new_name"
Session->>Session: 記錄屬性變更
App->>Session: session.flush()
Session->>UoW: 收集所有變更
UoW->>UoW: 拓撲排序
UoW->>DB: INSERT/UPDATE/DELETE
DB->>Session: 確認
2.3 Identity Map 模式
Identity Map 確保每個資料庫行在 Session 中只有一個對象實例:
flowchart TB
subgraph Session["Session"]
IM["Identity Map<br/>(表, 主鍵) → 對象"]
end
subgraph Objects["Python 對象"]
U1["User(id=1)"]
U2["User(id=2)"]
end
subgraph DB["資料庫"]
R1["Row: id=1"]
R2["Row: id=2"]
end
R1 -->|"查詢"| IM
R2 -->|"查詢"| IM
IM -->|"唯一映射"| U1
IM -->|"唯一映射"| U2
好處:
- 避免同一行的多個對象實例
- 減少資料庫查詢
- 保證對象狀態一致性
3. Session 與 Unit of Work 實現
3.1 Session 核心結構
# 簡化的 Session 結構
# 源自 orm/session.py
class Session:
def __init__(self, bind=None):
# Identity Map:存儲所有已載入的對象
self.identity_map = IdentityMap()
# 待處理的對象集合
self._new = set() # 新增對象
self._deleted = set() # 刪除對象
self._dirty = set() # 修改對象
# 資料庫連接
self.bind = bind
def add(self, instance):
"""將對象加入 Session"""
state = instance_state(instance)
self._new.add(state)
def delete(self, instance):
"""標記對象為刪除"""
state = instance_state(instance)
self._deleted.add(state)
def flush(self):
"""將所有變更寫入資料庫"""
# 創建 UOWTransaction
transaction = UOWTransaction(self)
# 處理所有變更
transaction.execute()
3.2 UOWTransaction 實現
# 源自 orm/unitofwork.py
class UOWTransaction:
"""Unit of Work 事務管理"""
def __init__(self, session: Session):
self.session = session
# mapper → 待處理的 InstanceState 集合
self.mappers = defaultdict(set)
# mapper → DependencyProcessor 集合
self.deps = defaultdict(set)
# 依賴關係圖(用於拓撲排序)
self.dependencies = set()
# 執行動作
self.postsort_actions = {}
def execute(self):
"""執行完整的 flush 流程"""
# 1. 收集所有待處理對象
self._collect_pending_states()
# 2. 計算依賴關係
self._calculate_dependencies()
# 3. 拓撲排序
sorted_actions = topological_sort(
self.postsort_actions,
self.dependencies
)
# 4. 按序執行
for action in sorted_actions:
action.execute()
3.3 Flush 流程詳解
flowchart TB
Start["session.flush()"]
subgraph Collect["1. 收集階段"]
CN["收集 _new 對象"]
CD["收集 _dirty 對象"]
CDE["收集 _deleted 對象"]
end
subgraph Preprocess["2. 預處理階段"]
CASCADE["處理 Cascade 規則"]
ORPHAN["處理孤兒對象"]
end
subgraph Sort["3. 排序階段"]
DEPS["計算表間依賴<br/>(外鍵關係)"]
TOPO["拓撲排序"]
end
subgraph Execute["4. 執行階段"]
INS["INSERT 操作"]
UPD["UPDATE 操作"]
DEL["DELETE 操作"]
end
subgraph Cleanup["5. 清理階段"]
RESET["重置對象狀態"]
REFRESH["刷新 Identity Map"]
end
Start --> CN
CN --> CD --> CDE
CDE --> CASCADE --> ORPHAN
ORPHAN --> DEPS --> TOPO
TOPO --> INS --> UPD --> DEL
DEL --> RESET --> REFRESH
3.4 拓撲排序算法
處理外鍵依賴的核心算法:
# 源自 util/topological.py
def topological_sort(actions, dependencies):
"""
拓撲排序,確保:
- 先 INSERT 被依賴的表
- 後 DELETE 被依賴的表
例如:
- users 表被 orders 引用
- INSERT: users → orders
- DELETE: orders → users
"""
# 構建入度圖
in_degree = defaultdict(int)
graph = defaultdict(list)
for action in actions:
in_degree[action] = 0
for (before, after) in dependencies:
graph[before].append(after)
in_degree[after] += 1
# BFS 拓撲排序
queue = deque([a for a in actions if in_degree[a] == 0])
result = []
while queue:
action = queue.popleft()
result.append(action)
for next_action in graph[action]:
in_degree[next_action] -= 1
if in_degree[next_action] == 0:
queue.append(next_action)
return result
4. Identity Map 實現
4.1 Identity Map 結構
# 源自 orm/identity.py
class IdentityMap:
"""Session 的 Identity Map 實現"""
def __init__(self):
# 核心字典:(Mapper, 主鍵) → InstanceState
self._dict: Dict[_IdentityKeyType, InstanceState] = {}
# 追蹤已修改的對象
self._modified: Set[InstanceState] = set()
def add(self, state: InstanceState) -> bool:
"""
添加對象到 Identity Map
如果已存在相同主鍵的對象,拋出異常
"""
key = state.key # (Mapper, 主鍵元組)
if key in self._dict:
existing = self._dict[key]
if existing is not state:
raise InvalidRequestError(
f"另一個相同主鍵的實例已存在"
)
return False # 已存在
self._dict[key] = state
self._manage_incoming_state(state)
return True
def __getitem__(self, key: _IdentityKeyType) -> object:
"""根據 Identity Key 獲取對象"""
state = self._dict[key]
return state.obj() # 弱引用解引用
def get(self, key, default=None):
"""安全獲取對象"""
try:
return self[key]
except KeyError:
return default
4.2 Identity Key 結構
# Identity Key 是一個元組
_IdentityKeyType = Tuple[
Type, # Mapper 類
Tuple[Any, ...] # 主鍵值元組
]
# 例如
key = (User, (42,)) # User 類,主鍵 id=42
key = (Order, (1, 100)) # 複合主鍵
4.3 弱引用機制
Identity Map 使用 弱引用 避免內存洩漏:
class _WeakInstanceDict(IdentityMap):
"""使用弱引用的 Identity Map"""
def __getitem__(self, key):
state = self._dict[key]
obj = state.obj() # obj 是弱引用
if obj is None:
# 對象已被垃圾回收
raise KeyError(key)
return obj
def add(self, state):
# state.obj 是弱引用函數
# 當 Python 對象被垃圾回收時,
# Identity Map 中的條目會自動失效
self._dict[state.key] = state
flowchart LR
subgraph IdentityMap["Identity Map"]
KEY["(User, (42,))"]
STATE["InstanceState<br/>weakref → obj"]
end
subgraph Memory["Python 記憶體"]
OBJ["User 實例"]
end
KEY --> STATE
STATE -.->|"弱引用"| OBJ
5. InstanceState:對象狀態追蹤
5.1 InstanceState 結構
每個 ORM 對象都有一個關聯的 InstanceState:
# 源自 orm/state.py
class InstanceState:
"""追蹤單個 ORM 對象的狀態"""
__slots__ = (
"class_", # 對象的類
"manager", # ClassManager
"obj", # 弱引用到實際對象
"committed_state", # 提交時的屬性值
"expired_attributes", # 過期的屬性
)
# 對象狀態標誌
session_id: Optional[int] = None # 關聯的 Session
key: Optional[_IdentityKeyType] = None # Identity Key
insert_order: Optional[int] = None # 插入順序
# 狀態標誌
_pending: bool = False # 待插入
_deleted: bool = False # 待刪除
_detached: bool = False # 已分離
modified: bool = False # 已修改
@property
def persistent(self) -> bool:
"""對象是否持久化(在資料庫中)"""
return self.key is not None and self.session_id is not None
@property
def pending(self) -> bool:
"""對象是否待插入"""
return self._pending
@property
def transient(self) -> bool:
"""對象是否瞬態(未關聯 Session)"""
return self.key is None and self.session_id is None
5.2 對象生命週期狀態
stateDiagram-v2
[*] --> Transient: 創建對象
Transient --> Pending: session.add()
Pending --> Persistent: session.flush()
Persistent --> Deleted: session.delete() + flush()
Deleted --> Detached: session.expunge()
Persistent --> Detached: session.expunge()
Pending --> Transient: session.expunge()
Detached --> Pending: session.add()
Deleted --> [*]: 事務提交
note right of Transient: 未關聯 Session
note right of Pending: 在 Session 中,待 INSERT
note right of Persistent: 在 Session 和資料庫中
note right of Deleted: 待 DELETE
note right of Detached: 曾在 Session 中
5.3 狀態判斷邏輯
def object_state(obj):
"""獲取對象狀態"""
state = instance_state(obj)
if state.session_id is None:
if state.key is None:
return "transient"
else:
return "detached"
else:
if state._pending:
return "pending"
elif state._deleted:
return "deleted"
else:
return "persistent"
6. Mapper:類與表的映射
6.1 Mapper 核心結構
# 源自 orm/mapper.py
class Mapper:
"""
建立 Python 類與資料庫表的映射
核心職責:
- 定義列與屬性的對應
- 配置關係 (relationships)
- 處理繼承
- 管理屬性的儀器化 (instrumentation)
"""
def __init__(
self,
class_: Type, # Python 類
local_table: Table, # 資料庫表
properties: Dict = None, # 屬性映射
primary_key: List = None,
inherits: "Mapper" = None,
):
self.class_ = class_
self.local_table = local_table
self.inherits = inherits
# 列屬性
self._columntoproperty = {} # Column → ColumnProperty
self._props = {} # 屬性名 → MapperProperty
# 主鍵列
self.primary_key = primary_key or []
# 儀器化類
self._configure_class_instrumentation()
def _configure_class_instrumentation(self):
"""
對類進行儀器化:
- 替換 __init__
- 添加屬性描述器
- 設置事件監聽
"""
manager = ClassManager(self.class_)
manager.mapper = self
# 為每個屬性創建描述器
for key, prop in self._props.items():
prop.instrument_class(self)
6.2 聲明式映射
SQLAlchemy 2.0 推薦使用聲明式映射:
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[Optional[str]]
# 關係
orders: Mapped[List["Order"]] = relationship(back_populates="user")
這會在幕後創建:
flowchart TB
subgraph Declaration["聲明式類"]
USER["class User(Base)"]
COLS["id, name, email"]
RELS["orders relationship"]
end
subgraph Generated["生成的結構"]
MAPPER["Mapper[User]"]
TABLE["Table('users')"]
PROPS["Properties:<br/>- ColumnProperty(id)<br/>- ColumnProperty(name)<br/>- RelationshipProperty(orders)"]
end
subgraph Instrumented["儀器化"]
MGR["ClassManager"]
ATTRS["InstrumentedAttribute<br/>描述器"]
end
USER --> MAPPER
COLS --> TABLE
MAPPER --> TABLE
MAPPER --> PROPS
MAPPER --> MGR
MGR --> ATTRS
6.3 MapperProperty 類型
| 類型 | 描述 | 範例 |
|---|---|---|
ColumnProperty | 列映射 | id, name |
RelationshipProperty | 關係映射 | user.orders |
SynonymProperty | 別名 | synonym('name') |
CompositeProperty | 複合屬性 | 多列組合 |
7. 屬性追蹤系統
7.1 InstrumentedAttribute
SQLAlchemy 使用 描述器協議 追蹤屬性變更:
# 源自 orm/attributes.py
class InstrumentedAttribute:
"""
Python 描述器,攔截屬性訪問
當訪問 user.name 時:
1. __get__ 返回實際值(或觸發 lazy load)
2. __set__ 記錄變更歷史
"""
def __init__(self, class_, key, impl):
self.class_ = class_
self.key = key
self.impl = impl # _AttributeImpl
def __get__(self, instance, owner):
if instance is None:
return self # 類級別訪問
state = instance_state(instance)
return self.impl.get(state, instance.__dict__)
def __set__(self, instance, value):
state = instance_state(instance)
old_value = self.impl.get(state, instance.__dict__)
# 記錄變更歷史
if old_value is not value:
state._commit_attr(self.key, old_value)
state.modified = True
# 觸發事件
self.impl.set(state, instance.__dict__, value)
def __delete__(self, instance):
state = instance_state(instance)
self.impl.delete(state, instance.__dict__)
7.2 變更歷史追蹤
class History:
"""屬性的變更歷史"""
__slots__ = ("added", "unchanged", "deleted")
def __init__(self, added, unchanged, deleted):
self.added = added # 新增值
self.unchanged = unchanged # 未變值
self.deleted = deleted # 刪除值
@classmethod
def from_scalar(cls, old, new):
if old is None and new is not None:
return cls([new], [], [])
elif old is not None and new is None:
return cls([], [], [old])
elif old != new:
return cls([new], [], [old])
else:
return cls([], [old], [])
7.3 髒檢查機制
sequenceDiagram
participant App as 應用
participant Attr as InstrumentedAttribute
participant State as InstanceState
participant Session as Session
App->>Attr: user.name = "new_name"
Attr->>Attr: __set__ 觸發
Attr->>State: 記錄舊值 committed_state
Attr->>State: modified = True
State->>Session: 加入 _dirty 集合
Note over Session: flush() 時
Session->>State: 比較 committed_state
Session->>Session: 生成 UPDATE SQL
8. 關係載入策略
8.1 載入策略類型
| 策略 | 描述 | 適用場景 |
|---|---|---|
| Lazy Loading | 訪問時載入 | 默認,少量訪問 |
| Eager Loading | 查詢時一起載入 | 已知需要關聯資料 |
| Joined Loading | JOIN 查詢 | 一對一/多對一 |
| Subquery Loading | 子查詢 | 一對多 |
| Select IN Loading | IN 查詢 | 大量關聯 |
8.2 策略實現
# 源自 orm/strategies.py
class LazyLoader(AbstractRelationshipLoader):
"""延遲載入策略"""
def create_row_processor(self, context, query_entity, path, mapper):
def load_lazy(state, passive):
# 訪問關係時才執行查詢
session = state.session
if session is None:
raise DetachedInstanceError()
# 構建查詢
query = session.query(self.mapper.class_)
query = query.filter(
self.mapper.primary_key == state.key[1]
)
return query.all()
return load_lazy
class JoinedLoader(AbstractRelationshipLoader):
"""JOIN 載入策略"""
def setup_query(self, context, query_entity, path, mapper):
# 添加 JOIN 到查詢
query = context.query
query = query.outerjoin(
self.mapper.local_table,
self.parent_property.primaryjoin
)
# 添加需要的列
query = query.add_columns(*self.mapper.columns)
return query
8.3 策略選擇流程
flowchart TB
QUERY["Query: session.query(User)"]
subgraph Strategies["載入策略"]
LAZY["lazy='select'<br/>(延遲載入)"]
JOINED["lazy='joined'<br/>(JOIN 載入)"]
SUBQ["lazy='subquery'<br/>(子查詢載入)"]
SELECTIN["lazy='selectin'<br/>(SELECT IN 載入)"]
end
subgraph Execution["執行"]
Q1["SELECT * FROM users"]
Q2["SELECT * FROM users<br/>JOIN orders ON ..."]
Q3["SELECT * FROM orders<br/>WHERE user_id IN<br/>(SELECT id FROM ...)"]
Q4["SELECT * FROM orders<br/>WHERE user_id IN (1,2,3,4)"]
end
QUERY --> LAZY --> Q1
QUERY --> JOINED --> Q2
QUERY --> SUBQ --> Q3
QUERY --> SELECTIN --> Q4
Q1 -.->|"訪問 user.orders"| Q4
8.4 動態切換策略
from sqlalchemy.orm import joinedload, selectinload
# 動態指定載入策略
session.query(User).options(
joinedload(User.orders), # 這次用 JOIN
selectinload(User.addresses) # 這次用 SELECT IN
)
9. Query 與 SQL 生成
9.1 Query 對象
# 源自 orm/query.py
class Query:
"""ORM 查詢構建器"""
def __init__(self, entities, session=None):
self._entities = entities # 查詢的實體
self.session = session
self._criterion = [] # WHERE 條件
self._order_by = [] # ORDER BY
self._limit = None
self._offset = None
def filter(self, *criterion):
"""添加 WHERE 條件"""
q = self._clone()
q._criterion.extend(criterion)
return q
def order_by(self, *columns):
"""添加 ORDER BY"""
q = self._clone()
q._order_by.extend(columns)
return q
def all(self):
"""執行查詢,返回所有結果"""
context = self._compile_context()
result = self.session.execute(context.statement)
return self._instances(result)
9.2 SQLAlchemy 2.0 風格
from sqlalchemy import select
# 新的 select() 語法
stmt = (
select(User)
.where(User.name == "John")
.order_by(User.created_at.desc())
)
result = session.execute(stmt)
users = result.scalars().all()
9.3 SQL 生成過程
flowchart TB
subgraph Python["Python 表達式"]
EXPR["User.name == 'John'"]
end
subgraph AST["SQL AST"]
CMP["BinaryExpression<br/>operator: =="]
COL["Column('name')"]
VAL["BindParameter('John')"]
CMP --> COL
CMP --> VAL
end
subgraph Compile["編譯"]
DIALECT["PostgreSQL Dialect"]
SQL["SELECT users.id, users.name<br/>FROM users<br/>WHERE users.name = $1"]
end
EXPR --> CMP
AST --> DIALECT
DIALECT --> SQL
10. 事件系統
10.1 事件類型
SQLAlchemy 提供豐富的事件 hook:
| 事件類別 | 範例 | 用途 |
|---|---|---|
| Mapper 事件 | before_insert, after_update | 持久化前後 |
| Session 事件 | after_flush, after_commit | Session 生命週期 |
| Attribute 事件 | set, append, remove | 屬性變更 |
| Instance 事件 | load, refresh | 對象載入 |
10.2 事件監聽
from sqlalchemy import event
# 監聽 INSERT 事件
@event.listens_for(User, "before_insert")
def before_insert_user(mapper, connection, target):
target.created_at = datetime.utcnow()
# 監聯 Session 事件
@event.listens_for(Session, "after_flush")
def after_flush(session, flush_context):
print(f"Flushed {len(session.new)} new objects")
# 監聽屬性變更
@event.listens_for(User.email, "set")
def validate_email(target, value, oldvalue, initiator):
if value and "@" not in value:
raise ValueError("Invalid email")
return value
11. 完整操作流程
11.1 CRUD 操作流程圖
sequenceDiagram
participant App as 應用程式
participant Session as Session
participant IM as Identity Map
participant UoW as Unit of Work
participant DB as 資料庫
Note over App,DB: === CREATE ===
App->>App: user = User(name="John")
App->>Session: session.add(user)
Session->>Session: _new.add(state)
Note over App,DB: === READ ===
App->>Session: session.get(User, 1)
Session->>IM: 查詢 (User, (1,))
alt 在 Identity Map 中
IM->>Session: 返回現有對象
else 不在
Session->>DB: SELECT * FROM users WHERE id = 1
DB->>Session: Row
Session->>Session: 創建 User 實例
Session->>IM: 添加到 Identity Map
end
Session->>App: 返回 User 對象
Note over App,DB: === UPDATE ===
App->>App: user.name = "Jane"
App->>Session: (自動追蹤)
Session->>Session: _dirty.add(state)
Note over App,DB: === DELETE ===
App->>Session: session.delete(user)
Session->>Session: _deleted.add(state)
Note over App,DB: === FLUSH ===
App->>Session: session.flush()
Session->>UoW: 收集變更
UoW->>UoW: 拓撲排序
UoW->>DB: INSERT/UPDATE/DELETE
DB->>Session: 確認
Session->>IM: 更新狀態
Note over App,DB: === COMMIT ===
App->>Session: session.commit()
Session->>DB: COMMIT
Session->>Session: 清理暫態
12. 性能優化建議
12.1 常見陷阱
| 問題 | 原因 | 解決方案 |
|---|---|---|
| N+1 查詢 | 延遲載入關係 | 使用 joinedload / selectinload |
| 大量對象 | Session 累積 | 定期 session.expire_all() |
| 批量操作慢 | 逐條 INSERT | 使用 bulk_insert_mappings |
| 長事務 | 未及時 commit | 縮短事務範圍 |
12.2 批量操作
# 低效:逐條插入
for data in large_data:
session.add(User(**data))
session.commit()
# 高效:批量插入
session.execute(
insert(User),
[{"name": d["name"]} for d in large_data]
)
session.commit()
12.3 Session 管理
# 推薦:使用 context manager
with Session(engine) as session:
with session.begin():
user = session.get(User, 1)
user.name = "Updated"
# 自動 commit 或 rollback
13. 總結
SQLAlchemy 核心設計
| 模式 | 實現 | 作用 |
|---|---|---|
| Data Mapper | Mapper | 類與表解耦 |
| Unit of Work | UOWTransaction | 變更追蹤與批量持久化 |
| Identity Map | IdentityMap | 對象唯一性保證 |
| Lazy Loading | LoaderStrategies | 按需載入 |