Coverage for apps/outers/repositories/session_repository.py: 57%

1from uuid import UUID 

2 

3import sqlalchemy 

4from sqlalchemy import exc 

5from sqlalchemy.engine import ScalarResult 

6from sqlmodel import select 

7from sqlmodel.ext.asyncio.session import AsyncSession 

8 

9from apps.inners.exceptions import repository_exception 

10from apps.inners.models.daos.session import Session 

11 

12 

13class SessionRepository: 

14 

15 def __init__(self): 

16 pass 

17 

18 async def find_one_by_id(self, session: AsyncSession, id: UUID) -> Session: 

19 try: 

20 found_session_result: ScalarResult = await session.exec( 

21 select(Session).where(Session.id == id).limit(1) 

22 ) 

23 found_session: Session = found_session_result.one() 

24 except sqlalchemy.exc.NoResultFound: 

25 raise repository_exception.NotFound() 

26 

27 return found_session 

28 

29 async def find_one_by_account_id(self, session: AsyncSession, account_id: UUID) -> Session: 

30 try: 

31 found_session_result: ScalarResult = await session.exec( 

32 select(Session).where(Session.account_id == account_id).limit(1) 

33 ) 

34 found_session: Session = found_session_result.one() 

35 except sqlalchemy.exc.NoResultFound: 

36 raise repository_exception.NotFound() 

37 

38 return found_session 

39 

40 async def find_one_by_access_token(self, session: AsyncSession, access_token: str) -> Session: 

41 try: 

42 found_session_result: ScalarResult = await session.exec( 

43 select(Session).where(Session.access_token == access_token).limit(1) 

44 ) 

45 found_session: Session = found_session_result.one() 

46 except sqlalchemy.exc.NoResultFound: 

47 raise repository_exception.NotFound() 

48 

49 return found_session 

50 

51 async def find_one_by_refresh_token(self, session: AsyncSession, refresh_token: str) -> Session: 

52 try: 

53 found_session_result: ScalarResult = await session.exec( 

54 select(Session).where(Session.refresh_token == refresh_token).limit(1) 

55 ) 

56 found_session: Session = found_session_result.one() 

57 except sqlalchemy.exc.NoResultFound: 

58 raise repository_exception.NotFound() 

59 

60 return found_session 

61 

62 def create_one(self, session: AsyncSession, session_creator: Session) -> Session: 

63 try: 

64 session.add(session_creator) 

65 except sqlalchemy.exc.IntegrityError: 

66 raise repository_exception.IntegrityError() 

67 

68 return session_creator 

69 

70 async def patch_one_by_id(self, session: AsyncSession, id: UUID, session_patcher: Session) -> Session: 

71 found_session: Session = await self.find_one_by_id( 

72 session=session, 

73 id=id 

74 ) 

75 found_session.patch_from(session_patcher.dict(exclude_none=True)) 

76 

77 return found_session 

78 

79 async def delete_one_by_id(self, session: AsyncSession, id: UUID) -> Session: 

80 found_session: Session = await self.find_one_by_id( 

81 session=session, 

82 id=id 

83 ) 

84 await session.delete(found_session) 

85 

86 return found_session