Coverage for app/ddd/infrastructure/repository/user_repository_impl.py: 100%

1# class ErrorCode: 

2 

3from sqlmodel import Session, select 

4 

5from app.ddd.domain import ( 

6 User, 

7 UserDuplicationError, 

8 UserId, 

9 UserNotFoundError, 

10 UserRepository, 

11 UserUpdateConflictError, 

12 # UserReportDuplicationError, 

13) 

14from migrations.models import TUser, TUserReport 

15 

16 

17class UserRepositoryImpl(UserRepository): 

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

19 self.__session: Session = session 

20 

21 def _fetch_by_id(self, _id: str) -> TUser | None: 

22 statement = select(TUser).where(TUser.user_id == _id) 

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

24 

25 def _fetch_user_report_by_id(self, _id: str) -> TUserReport | None: 

26 statement = select(TUserReport).where(TUserReport.user_report_id == _id) 

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

28 

29 def _apply(self, model: TUser | TUserReport) -> None: 

30 self.__session.add(model) 

31 

32 def _delete(self, model: TUser) -> None: 

33 self.__session.delete(model) 

34 

35 def find_by_id(self, _id: UserId) -> User: 

36 model: TUser | None = self._fetch_by_id(_id.root) 

37 if model is None: 

38 raise UserNotFoundError(_id.root) 

39 return User.model_validate(model) 

40 

41 def insert(self, user: User) -> None: 

42 model: TUser | None = self._fetch_by_id(user.user_id) 

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

44 raise UserDuplicationError(user.user_id) 

45 model = TUser.model_validate(user.model_dump(exclude={"created_at", "updated_at"})) # 新規作成 

46 self._apply(model) 

47 

48 def insert_user_report(self, user: User) -> None: 

49 for user_report in user.user_reports: 

50 _id = user_report.user_report_id 

51 model: TUserReport | None = self._fetch_user_report_by_id(_id) 

52 print(_id) 

53 print(model) 

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

55 # raise UserReportDuplicationError(_id) 

56 continue 

57 model = TUserReport.model_validate(user_report.model_dump(exclude={"created_at", "updated_at"})) # 新規作成 

58 self._apply(model) 

59 

60 def update(self, user: User) -> None: 

61 model: TUser | None = self._fetch_by_id(user.user_id) 

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

63 raise UserNotFoundError(user.user_id) 

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

65 raise UserUpdateConflictError(user_id=user.user_id) 

66 model.sqlmodel_update(user.model_dump(exclude={"created_at", "updated_at", "user_reports"})) # 更新データの統合 

67 self._apply(model) 

68 

69 def delete(self, _id: UserId) -> None: 

70 model: TUser | None = self._fetch_by_id(_id.root) 

71 if model is None: 

72 raise UserNotFoundError(_id.root) 

73 

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

75 # model.updated_at = datetime.now() 

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

77 # self._apply(model) 

78 

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

80 

81 def query(self) -> list[User]: 

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

83 models: list[TUser] = self.__session.query(TUser).all() 

84 return [User.model_validate(model) for model in models] 

85