Coverage for apps/outers/repositories/account_repository.py: 64%

1from typing import List 

2from uuid import UUID 

3 

4import sqlalchemy 

5from sqlalchemy import exc 

6from sqlalchemy.engine import ScalarResult 

7from sqlmodel import select 

8from sqlmodel.ext.asyncio.session import AsyncSession 

9 

10from apps.inners.exceptions import repository_exception 

11from apps.inners.models.daos.account import Account 

12 

13 

14class AccountRepository: 

15 

16 def __init__(self): 

17 pass 

18 

19 async def find_many_by_account_id_with_pagination( 

20 self, 

21 session: AsyncSession, 

22 account_id: UUID, 

23 page_position: int, 

24 page_size: int 

25 ) -> List[Account]: 

26 found_account_result: ScalarResult = await session.exec( 

27 select(Account) 

28 .where(Account.id == account_id) 

29 .limit(page_size) 

30 .offset(page_size * (page_position - 1)) 

31 ) 

32 found_accounts: List[Account] = list(found_account_result.all()) 

33 

34 return found_accounts 

35 

36 async def find_one_by_id_and_account_id(self, session: AsyncSession, id: UUID, account_id: UUID) -> Account: 

37 try: 

38 found_account_result: ScalarResult = await session.exec( 

39 select(Account) 

40 .where(Account.id == id) 

41 .where(Account.id == account_id) 

42 .limit(1) 

43 ) 

44 found_account: Account = found_account_result.one() 

45 except sqlalchemy.exc.NoResultFound: 

46 raise repository_exception.NotFound() 

47 

48 return found_account 

49 

50 async def find_one_by_email(self, session: AsyncSession, email: str) -> Account: 

51 try: 

52 found_account_result: ScalarResult = await session.exec( 

53 select(Account).where(Account.email == email).limit(1) 

54 ) 

55 found_account: Account = found_account_result.one() 

56 except sqlalchemy.exc.NoResultFound: 

57 raise repository_exception.NotFound() 

58 

59 return found_account 

60 

61 def create_one(self, session: AsyncSession, account_creator: Account) -> Account: 

62 try: 

63 session.add(account_creator) 

64 except sqlalchemy.exc.IntegrityError: 

65 raise repository_exception.IntegrityError() 

66 

67 return account_creator 

68 

69 async def patch_one_by_id_and_account_id( 

70 self, 

71 session: AsyncSession, 

72 id: UUID, 

73 account_id: UUID, 

74 account_patcher: Account 

75 ) -> Account: 

76 found_account: Account = await self.find_one_by_id_and_account_id( 

77 session=session, 

78 id=id, 

79 account_id=account_id 

80 ) 

81 found_account.patch_from(account_patcher.dict(exclude_none=True)) 

82 

83 return found_account 

84 

85 async def delete_one_by_id_and_account_id(self, session: AsyncSession, id: UUID, account_id: UUID) -> Account: 

86 found_account: Account = await self.find_one_by_id_and_account_id( 

87 session=session, 

88 id=id, 

89 account_id=account_id 

90 ) 

91 await session.delete(found_account) 

92 return found_account