Coverage for app/ddd/infrastructure/repository/group_repository_impl.py: 75%

1# class ErrorCode: 

2 

3from sqlmodel import Session, select 

4 

5from app.ddd.domain import ( 

6 Group, 

7 GroupDuplicationError, 

8 GroupId, 

9 GroupNotFoundError, 

10 GroupRepository, 

11 GroupUpdateConflictError, 

12) 

13from migrations.models import TGroup 

14 

15 

16class GroupRepositoryImpl(GroupRepository): 

17 def __init__(self, session: Session) -> None: 

18 self.__session: Session = session 

19 

20 def _fetch_by_id(self, _id: str) -> TGroup | None: 

21 statement = select(TGroup).where(TGroup.group_id == _id) 

22 return self.__session.exec(statement).first() 

23 

24 def _apply(self, model: TGroup) -> None: 

25 self.__session.add(model) 

26 

27 def _delete(self, model: TGroup) -> None: 

28 self.__session.delete(model) 

29 

30 def find_by_id(self, _id: GroupId) -> Group: 

31 model: TGroup | None = self._fetch_by_id(_id.root) 

32 if model is None: 

33 raise GroupNotFoundError(_id.root) 

34 return Group.model_validate(model) 

35 

36 def insert(self, group: Group) -> None: 

37 model: TGroup | None = self._fetch_by_id(group.group_id) 

38 if model is not None: # 重複判定 

39 raise GroupDuplicationError(group.group_id) 

40 model = TGroup.model_validate(group.model_dump(exclude={"created_at", "updated_at"})) # 新規作成 

41 self._apply(model) 

42 

43 

44 def update(self, group: Group) -> None: 

45 model: TGroup | None = self._fetch_by_id(group.group_id) 

46 if model is None: # ない場合更新できない 現在のユースケースでは基本的に発生しない 

47 raise GroupNotFoundError(group.group_id) 

48 if group.updated_at != model.updated_at: # 更新日検証による楽観的ロックの確認 

49 raise GroupUpdateConflictError(group_id=group.group_id) 

50 model.sqlmodel_update(group.model_dump(exclude={"created_at", "updated_at", "group_reports"})) # 更新データの統合 

51 self._apply(model) 

52 

53 def delete(self, _id: GroupId) -> None: 

54 model: TGroup | None = self._fetch_by_id(_id.root) 

55 if model is None: 

56 raise GroupNotFoundError(_id.root) 

57 

58 # TODO(nonomura): 論理削除する場合 

59 # model.updated_at = datetime.now() 

60 # model.deleted_at = datetime.now() # 論理削除する場合 

61 # self._apply(model) 

62 

63 self._delete(model) # 物理削除用 

64 

65 def query(self) -> list[Group]: 

66 """TODO(nonomura): 検索条件整理後に整理。必要に応じてクエリサービスに分ける.""" 

67 models: list[TGroup] = self.__session.query(TGroup).all() 

68 return [Group.model_validate(model) for model in models] 

69