Python技术迷

一次构建,随处复用:Python 中的泛型仓库模式

接口跑得好好的,改了个字段类型,仓库层先炸了。

日志不复杂,就一句:

AttributeError: 'OrderRepo' object has no attribute 'find_by_phone'

这类问题我第一眼一般不看 SQL,也不看 ORM,先看仓库类有没有开始“长野了”。十有八九是 UserRepository、OrderRepository、SkuRepository 各写一套,写着写着就分叉了。

比如项目里常见这种代码:

classOrderRepo:
def__init__(self, db):
        self.db = db

defget(self, order_id: int):
return self.db.orders.get(order_id)

defsave(self, order):
        self.db.orders[order.id] = order
return order


classUserRepo:
def__init__(self, db):
        self.db = db

defget(self, user_id: int):
return self.db.users.get(user_id)

defsave(self, user):
        self.db.users[user.id] = user
return user

看着没毛病,实际很烦。

get、save、delete 这些动作每个仓库都要写一遍。写到第三个表的时候,人就开始偷懒。某个仓库返回 None,某个仓库抛异常,某个仓库保存前做校验,另一个忘了做。

面试里问“泛型仓库模式”,不是想听你背 Generic、TypeVar 的定义。

他想看你能不能把这种重复的仓库代码收住,同时类型还别丢。

Python 里可以这么写。

from dataclasses import dataclass
from typing import Generic, TypeVar, Protocol

Id = TypeVar("Id")


classHasId(Protocol[Id]):
    id: Id


T = TypeVar("T", bound=HasId)


classMemoryRepo(Generic[Id, T]):
def__init__(self) -> None:
        self._rows: dict[Id, T] = {}

defget(self, row_id: Id) -> T | None:
return self._rows.get(row_id)

defmust_get(self, row_id: Id) -> T:
        row = self.get(row_id)
if row isNone:
raise LookupError(f"row not found, id={row_id}")
return row

defsave(self, row: T) -> T:
        self._rows[row.id] = row
return row

defdelete(self, row_id: Id) -> bool:
return self._rows.pop(row_id, None) isnotNone

deflist_all(self) -> list[T]:
return list(self._rows.values())

这里我会用 Protocol,不喜欢把基类焊死。

很多人会写一个 BaseEntity,然后要求所有实体都继承它。这个味儿有点重,后面一接第三方对象、ORM 对象、历史老对象,就开始补胶水代码。

Protocol 的意思更松一点:我不管你继承谁,只要你有 id 字段,就能进这个仓库。

业务对象可以很干净:

@dataclass
classOrder:
    id: str
    user_id: int
    amount: int
    status: str


@dataclass
classUser:
    id: int
    name: str
    locked: bool = False

然后仓库这么用:

order_repo = MemoryRepo[str, Order]()
user_repo = MemoryRepo[int, User]()

order_repo.save(Order(id="O20260424001", user_id=7, amount=399, status="PAID"))
user_repo.save(User(id=7, name="rabbit"))

order = order_repo.must_get("O20260424001")
user = user_repo.must_get(7)

这就是“一次构建,随处复用”。

同一套 MemoryRepo,订单能用,用户能用,商品也能用。关键是类型没丢。

如果你写错 ID 类型,类型检查工具会直接提醒:

order_repo.must_get(1001)

订单 ID 明明是 str,你传了个 int。这要是在普通 Python 代码里,运行前不一定看得出来。线上查不到数据,日志里只剩一句“订单不存在”,排查半天,最后发现是 ID 类型传错了。

这种问题不高级,但恶心。

不过泛型仓库也别写过头。

我见过有人把仓库抽成这样:

Repository[T, ID, Query, Command, Result, Context]

第一眼就不太信。

仓库层不是论文。抽象到开发同事每次写业务都得猜泛型参数,那这个抽象基本已经开始反噬了。

比较稳的做法是:通用 CRUD 放基类,业务查询留给具体仓库。

比如订单仓库可以继承一下:

classOrderRepo(MemoryRepo[str, Order]):
deflist_paid_by_user(self, user_id: int) -> list[Order]:
        rows = []
for order in self.list_all():
if order.user_id == user_id and order.status == "PAID":
                rows.append(order)
return rows

这段看着普通,但边界是清楚的。

save、get、delete 这种动作,不应该每个仓库重复写。

list_paid_by_user 这种带业务含义的查询,就别硬塞进泛型仓库里。你要是为了“通用”,搞一个 find_by(field, value),刚开始很爽,后面字段名写错、条件组合变复杂、排序分页加进来,就会变成半个蹩脚 ORM。

如果是数据库场景,泛型仓库的核心也差不多,只是底层从 dict 换成 session。

示意一下,不依赖具体 ORM:

classSqlRepo(Generic[Id, T]):
def__init__(self, session, model: type[T]) -> None:
        self.session = session
        self.model = model

defget(self, row_id: Id) -> T | None:
return self.session.get(self.model, row_id)

defsave(self, row: T) -> T:
        self.session.add(row)
return row

defdelete(self, row_id: Id) -> bool:
        row = self.get(row_id)
if row isNone:
returnFalse
        self.session.delete(row)
returnTrue

这里有个坑要注意。

仓库里不要偷偷 commit。

我一般不让 save() 自己提交事务。仓库只负责把对象放进 session,什么时候提交交给 service 层。否则一个接口里保存订单、扣库存、写流水,前两个仓库各自 commit,第三步失败了,你回滚都回不干净。

更像样一点的调用应该是这样:

defconfirm_order(order_id: str, repo: OrderRepo, tx) -> None:
    order = repo.must_get(order_id)

if order.status != "NEW":
raise ValueError(f"bad order status: {order.status}")

    order.status = "PAID"
    repo.save(order)

    tx.commit()

仓库模式解决的是“数据访问动作重复”和“业务代码直接摸存储细节”的问题。

泛型解决的是“同一套仓库逻辑复用以后,类型还能跟着走”的问题。

别把它想得太玄。面试时能把这三点说清楚,再手写一段 Generic + TypeVar + Protocol,基本就够了:

通用仓库放公共动作。

具体仓库放业务查询。

事务边界不要藏在仓库里。

这种设计不花哨,但项目长一点以后,你会感谢自己当时没把每个 Repository 都复制一遍。