Repository và ORM: biên nào che một quyết định thật?

Câu hỏi: thêm OrderStore giúp giữ luật hủy đơn hay chỉ thêm một bước gọi?

Cần biết trước: module/domain, transaction và Python. Lab dùng Python3.14, SQLAlchemy2.0.54, SQLite trong thư mục thử riêng. Đây là phép kiểm hành vi và số câu SQL, không đo tốc độ ORM hoặc concurrency PostgreSQL.

So cùng bài toán

Order SHIPPED không hủy được, lý do không trắng; CANCELLED hủy lại là no-op. Trạng thái và sự kiện hủy phải commit cùng transaction. Direct dùng ORM ngay trong use case; phương án có biên trả Order thuần và adapter lưu nó. Cả hai dùng cùng luật. Read projection đếm theo trạng thái dùng SQL set-based, không ép nạp từng aggregate.

Session quản lý transaction và identity map; context transaction commit khi thành công, rollback khi exception. Repository không tự có cơ chế đó: ở lab, caller giữ transaction cho cả save và event. Session.

Lưu bốn file và runner trong thư mục trống

File thứ nhất là domain thuần. Hủy lặp giữ lý do đầu tiên trong event đã lưu.

from __future__ import annotations

from dataclasses import dataclass


def cancel_status(status: str, reason: str) -> str:
    if status == "CANCELLED":
        return status
    if status == "SHIPPED" or not reason.strip():
        raise ValueError("không được hủy")
    return "CANCELLED"


@dataclass(slots=True)
class Order:
    id: int
    status: str

    def cancel(self, reason: str) -> None:
        self.status = cancel_status(self.status, reason)

from __future__ import annotations

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class OrderRow(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str] = mapped_column("status")


class CancelEvent(Base):
    __tablename__ = "cancel_events"
    order_id: Mapped[int] = mapped_column(primary_key=True)
    reason: Mapped[str]

from __future__ import annotations

from mapping import OrderRow
from sqlalchemy import func, select
from sqlalchemy.orm import Session


def summary(session: Session) -> dict[str, int]:
    query = select(OrderRow.status, func.count()).group_by(OrderRow.status)
    return {status: count for status, count in session.execute(query)}

Biên sau che cách nạp/lưu domain; caller vẫn biết transaction là một Session. Đây là demo nhỏ, chưa tách UnitOfWork port. Không nói use case đã độc lập mọi ORM.

from __future__ import annotations

from typing import Protocol

from mapping import OrderRow
from order import Order
from sqlalchemy.orm import Session


class OrderStore(Protocol):
    def get(self, order_id: int) -> Order: ...
    def save(self, order: Order) -> None: ...


class SqlOrderStore:
    def __init__(self, session: Session) -> None:
        self.session = session
        self.loaded: dict[int, OrderRow] = {}

    def get(self, order_id: int) -> Order:
        row = self.session.get(OrderRow, order_id)
        if row is None:
            raise LookupError(order_id)
        self.loaded[order_id] = row
        return Order(row.id, row.status)

    def save(self, order: Order) -> None:
        self.loaded[order.id].status = order.status

Runner so hai use case trên database mới, không dùng fake repository thay transaction. Mỗi transaction có một adapter mới; cache loaded chỉ sống trong transaction ấy.

from __future__ import annotations

import hashlib
import importlib.metadata
import sqlite3
import subprocess
import sys
from pathlib import Path

from mapping import Base, CancelEvent, OrderRow
from order import cancel_status
from queries import summary
from sqlalchemy import create_engine, event, func, select
from sqlalchemy.orm import Session
from store import OrderStore, SqlOrderStore


def direct(session: Session, order_id: int, reason: str) -> None:
    row = session.get(OrderRow, order_id)
    assert row is not None
    before = row.status
    row.status = cancel_status(row.status, reason)
    if before != row.status:
        session.add(CancelEvent(order_id=order_id, reason=reason.strip()))


def boundary(session: Session, store: OrderStore, order_id: int, reason: str) -> None:
    order = store.get(order_id)
    before = order.status
    order.cancel(reason)
    store.save(order)
    if before != order.status:
        session.add(CancelEvent(order_id=order_id, reason=reason.strip()))


def exercise() -> None:
    assert importlib.metadata.version("SQLAlchemy") == "2.0.54"
    for mode in ("direct", "repo"):
        engine = create_engine(f"sqlite:///{mode}.db")
        statements: list[str] = []
        event.listen(
            engine,
            "before_cursor_execute",
            lambda c, u, q, p, x, m, log=statements: log.append(q),
        )
        Base.metadata.create_all(engine)
        try:
            with Session(engine) as session, session.begin():
                session.add_all(
                    [
                        OrderRow(id=1, status="CONFIRMED"),
                        OrderRow(id=2, status="SHIPPED"),
                    ]
                )
            for order_id, reason in ((2, "lý do"), (1, " ")):
                try:
                    with Session(engine) as session, session.begin():
                        if mode == "direct":
                            direct(session, order_id, reason)
                        else:
                            boundary(session, SqlOrderStore(session), order_id, reason)
                except ValueError:
                    pass
                else:
                    raise AssertionError("invariant không được bảo vệ")
            try:
                with Session(engine) as session, session.begin():
                    if mode == "direct":
                        direct(session, 1, "first")
                    else:
                        boundary(session, SqlOrderStore(session), 1, "first")
                    session.flush()
                    raise RuntimeError("fault sau flush")
            except RuntimeError as error:
                assert str(error) == "fault sau flush"
            with Session(engine) as session:
                row = session.get(OrderRow, 1)
                assert row is not None and row.status == "CONFIRMED"
                assert (
                    session.scalar(select(func.count()).select_from(CancelEvent)) == 0
                )
            statements.clear()
            with Session(engine) as session, session.begin():
                if mode == "direct":
                    direct(session, 1, "first")
                else:
                    boundary(session, SqlOrderStore(session), 1, "first")
            assert len(statements) == 3, statements
            statements.clear()
            with Session(engine) as session, session.begin():
                if mode == "direct":
                    direct(session, 1, "second")
                else:
                    boundary(session, SqlOrderStore(session), 1, "second")
            assert len(statements) == 1
            with Session(engine) as session:
                assert (
                    session.scalar(select(func.count()).select_from(CancelEvent)) == 1
                )
                saved_event = session.get(CancelEvent, 1)
                assert saved_event is not None and saved_event.reason == "first"
                statements.clear()
                result = summary(session)
                expected = (
                    {"SHIPPED": 1}
                    if "--query" in sys.argv
                    else {"CANCELLED": 1, "SHIPPED": 1}
                )
                assert result == expected and len(statements) == 1
            sys.stdout.write(
                f"{mode}: invariant/rollback/noop=ok cancel_sql=3 repeat_sql=1 summary_sql=1\n"
            )
        finally:
            engine.dispose()
            Path(f"{mode}.db").unlink(missing_ok=True)


def main() -> None:
    exercise()
    if "--child" in sys.argv:
        return
    before = {
        p: hashlib.sha256(p.read_bytes()).hexdigest() for p in Path(".").glob("*.py")
    }
    mapping = Path("mapping.py")
    original = mapping.read_text()
    try:
        mapping.write_text(
            original.replace('mapped_column("status")', 'mapped_column("state")')
        )
        changed = [
            p.name
            for p, digest in before.items()
            if hashlib.sha256(p.read_bytes()).hexdigest() != digest
        ]
        assert changed == ["mapping.py"]
        subprocess.run(
            [sys.executable, "-B", "check.py", "--child"], check=True, timeout=30
        )
    finally:
        mapping.write_text(original)
    queries = Path("queries.py")
    original_query = queries.read_text()
    try:
        queries.write_text(
            original_query.replace(
                ".group_by(OrderRow.status)",
                '.where(OrderRow.status != "CANCELLED").group_by(OrderRow.status)',
            )
        )
        changed = [
            p.name
            for p, digest in before.items()
            if hashlib.sha256(p.read_bytes()).hexdigest() != digest
        ]
        assert changed == ["queries.py"]
        subprocess.run(
            [sys.executable, "-B", "check.py", "--child", "--query"],
            check=True,
            timeout=30,
        )
    finally:
        queries.write_text(original_query)
    sys.stdout.write(
        f"sqlite={sqlite3.sqlite_version} schema_change=1 query_change=1 both_paths=ok\n"
    )


if __name__ == "__main__":
    main()

Chạy và kiểm output

set -euo pipefail
uv run --no-project --with 'SQLAlchemy==2.0.54' python -B check.py

direct: invariant/rollback/noop=ok cancel_sql=3 repeat_sql=1 summary_sql=1
repo: invariant/rollback/noop=ok cancel_sql=3 repeat_sql=1 summary_sql=1

set -euo pipefail
test ! -e direct.db
test ! -e repo.db

Đọc phép so đúng phạm vi

Các biến thể tạo database mới để kiểm mapping, chưa chạy migration dữ liệu cũ. Runner so hash file thật và chạy lại cả hai đường; không đo số file trong mọi project. Mapped có thể dùng tên attribute khác tên cột database. Mapping API.

Repository hữu ích khi hợp đồng OrderStore là thứ application muốn giữ ổn định, domain không phụ thuộc tracking/query của ORM, hoặc cách lưu root đổi thật. Đổi schema vật lý đơn giản không tự tạo lợi ích: ORM cũng làm được như lab. Adapter ở đây thêm mapping/cache và interface, một chi phí cần nhận rõ.

Repository rỗng như get(id) -> session.get(Row,id) và trả Query/Row của chính ORM không che query/tracking hay domain model. Dùng ORM trực tiếp ở use case nhỏ có thể đủ, miễn invariant và transaction vẫn được kiểm. Ngược lại, repository “chung mọi bảng” có thể ép query đọc qua vòng lặp, gây nhiều SQL; kiểm SQL và dữ liệu ở N+1 thay vì tin tên pattern.

Lab chưa kiểm concurrent cancel, optimistic version hoặc database ngoài SQLite. Authorization phải kiểm trước hành vi hủy; idempotent không cấp quyền. Một transaction chỉ chứa database này, không bao gồm notification bên ngoài. Học tiếp: ADR, MVCC.