Coverage for apps/outers/repositories/web_document_repository.py: 74%

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.document import Document 

12from apps.inners.models.daos.web_document import WebDocument 

13 

14 

15class WebDocumentRepository: 

16 

17 def __init__(self): 

18 pass 

19 

20 async def find_many_by_account_id_with_pagination( 

21 self, 

22 session: AsyncSession, 

23 account_id: UUID, 

24 page_position: int, 

25 page_size: int 

26 ) -> List[WebDocument]: 

27 found_web_document_result: ScalarResult = await session.exec( 

28 select(WebDocument) 

29 .join(Document, Document.id == WebDocument.id) 

30 .where(Document.account_id == account_id) 

31 .limit(page_size) 

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

33 ) 

34 found_web_documents: List[WebDocument] = list(found_web_document_result.all()) 

35 

36 return found_web_documents 

37 

38 async def find_one_by_id_and_account_id(self, session: AsyncSession, id: UUID, account_id: UUID) -> WebDocument: 

39 try: 

40 found_web_document_result: ScalarResult = await session.exec( 

41 select(WebDocument) 

42 .join(Document, Document.id == WebDocument.id) 

43 .where(WebDocument.id == id) 

44 .where(Document.account_id == account_id) 

45 ) 

46 found_web_document: WebDocument = found_web_document_result.one() 

47 except sqlalchemy.exc.NoResultFound: 

48 raise repository_exception.NotFound() 

49 

50 return found_web_document 

51 

52 def create_one(self, session: AsyncSession, web_document_creator: WebDocument) -> WebDocument: 

53 try: 

54 session.add(web_document_creator) 

55 except sqlalchemy.exc.IntegrityError: 

56 raise repository_exception.IntegrityError() 

57 

58 return web_document_creator 

59 

60 async def patch_one_by_id_and_account_id( 

61 self, 

62 session: AsyncSession, 

63 id: UUID, 

64 account_id: UUID, 

65 web_document_patcher: WebDocument 

66 ) -> WebDocument: 

67 found_web_document: WebDocument = await self.find_one_by_id_and_account_id( 

68 session=session, 

69 id=id, 

70 account_id=account_id 

71 ) 

72 found_web_document.patch_from(web_document_patcher.dict(exclude_none=True)) 

73 

74 return found_web_document 

75 

76 async def delete_one_by_id_and_account_id(self, session: AsyncSession, id: UUID, account_id: UUID) -> WebDocument: 

77 found_web_document: WebDocument = await self.find_one_by_id_and_account_id( 

78 session=session, 

79 id=id, 

80 account_id=account_id 

81 ) 

82 await session.delete(found_web_document) 

83 

84 return found_web_document