Python工作单元模式

# Python 工作单元模式 (Unit of Work)# 跟踪变更批量提交事务管理异常回滚# 核心方法register_new / register_dirty / register_deleted。from abc import ABC, abstractmethodfrom dataclasses import dataclassfrom typing import Optionaldataclassclass Author:id: str; name: strclass AuthorRepository(ABC):abstractmethoddef add(self, a: Author) - None: ...abstractmethoddef update(self, a: Author) - None: ...abstractmethoddef delete(self, aid: str) - None: ...abstractmethoddef get(self, aid: str) - Optional[Author]: ...# 工作单元跟踪对象变更并在 commit 时一次性写入class UnitOfWork:def __init__(self):self._new: list[Author] []self._dirty: list[Author] []self._deleted: list[str] []def register_new(self, a: Author) - None:self._new.append(a)def register_dirty(self, a: Author) - None:if a not in self._dirty:self._dirty.append(a)def register_deleted(self, aid: str) - None:self._deleted.append(aid)def commit(self, repo: AuthorRepository) - None:try:for a in self._new:repo.add(a)for a in self._dirty:repo.update(a)for aid in self._deleted:repo.delete(aid)except Exception as e:raise RuntimeError(f提交失败: {e})finally:self._new.clear(); self._dirty.clear(); self._deleted.clear()class MemoryAuthorRepo(AuthorRepository):def __init__(self):self._store: dict[str, Author] {}def add(self, a: Author) - None:self._store[a.id] adef update(self, a: Author) - None:self._store[a.id] adef delete(self, aid: str) - None:self._store.pop(aid, None)def get(self, aid: str) - Optional[Author]:return self._store.get(aid)class AuthorService:def __init__(self, repo: AuthorRepository, uow: UnitOfWork):self._repo repo; self._uow uowdef create(self, aid: str, name: str) - None:self._uow.register_new(Author(aid, name))self._uow.commit(self._repo)def rename(self, aid: str, new_name: str) - None:a self._repo.get(aid)if a:a.name new_nameself._uow.register_dirty(a)self._uow.commit(self._repo)if __name__ __main__:repo MemoryAuthorRepo()uow UnitOfWork()svc AuthorService(repo, uow)svc.create(A001, 鲁迅)svc.rename(A001, 周树人)print(repo.get(A001))