一次构建,随处复用: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 都复制一遍。