SQLAlchemy ORM 原理深度解析:從 Unit of Work 到 Identity Map


0. 前言與生成說明

本文深入解析 Python 生態系統中最強大的 ORM 框架 —— SQLAlchemy 的內部實現原理。

文章目標

讀完本文,你將理解:

  • SQLAlchemy 的整體架構與設計理念
  • Session 與 Unit of Work 模式的實現
  • Identity Map 如何保證對象唯一性
  • Mapper 如何建立類與表的映射
  • 屬性追蹤與變更檢測機制
  • 關係載入策略的實現

1. SQLAlchemy 整體架構

SQLAlchemy 分為兩個主要層次:

層次描述核心組件
CoreSQL 表達式語言與資料庫抽象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領域對象不依賴 ORMSQLAlchemy
Active Record領域對象繼承 ORM 基類Django ORM、Rails

2.2 Unit of Work 模式

Unit of Work 是 SQLAlchemy 的核心模式,負責:

  1. 追蹤所有對象變更
  2. 計算變更順序(拓撲排序)
  3. 批量執行 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 LoadingJOIN 查詢一對一/多對一
Subquery Loading子查詢一對多
Select IN LoadingIN 查詢大量關聯

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_commitSession 生命週期
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 MapperMapper類與表解耦
Unit of WorkUOWTransaction變更追蹤與批量持久化
Identity MapIdentityMap對象唯一性保證
Lazy LoadingLoaderStrategies按需載入

學習資源


參考資料