Coverage for apps/inners/use_cases/managements/web_document_management.py: 73%

1import hashlib 

2import uuid 

3from typing import List 

4from uuid import UUID 

5 

6from starlette.datastructures import State 

7 

8from apps.inners.models.daos.document import Document 

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

10from apps.inners.models.dtos.constants.document_type_constant import DocumentTypeConstant 

11from apps.inners.models.dtos.contracts.requests.managements.web_documents.create_one_body import CreateOneBody 

12from apps.inners.models.dtos.contracts.requests.managements.web_documents.patch_one_body import PatchOneBody 

13from apps.inners.models.dtos.contracts.responses.managements.documents.web_document_response import WebDocumentResponse 

14from apps.inners.use_cases.managements.document_management import DocumentManagement 

15from apps.outers.repositories.web_document_repository import WebDocumentRepository 

16 

17 

18class WebDocumentManagement: 

19 def __init__( 

20 self, 

21 document_management: DocumentManagement, 

22 web_document_repository: WebDocumentRepository, 

23 ): 

24 self.document_management: DocumentManagement = document_management 

25 self.web_document_repository: WebDocumentRepository = web_document_repository 

26 

27 async def find_many_with_authorization_and_pagination(self, state: State, page_position: int, page_size: int) -> \ 

28 List[ 

29 WebDocumentResponse]: 

30 found_web_documents: List[ 

31 WebDocument 

32 ] = await self.web_document_repository.find_many_by_account_id_with_pagination( 

33 session=state.session, 

34 account_id=state.authorized_session.account_id, 

35 page_position=page_position, 

36 page_size=page_size 

37 ) 

38 found_documents: List[ 

39 Document 

40 ] = await self.document_management.document_repository.find_many_by_id_and_account_id( 

41 session=state.session, 

42 ids=[found_web_document.id for found_web_document in found_web_documents], 

43 account_id=state.authorized_session.account_id 

44 ) 

45 found_web_document_responses: List[WebDocumentResponse] = [] 

46 for found_document, found_web_document in zip(found_documents, found_web_documents, strict=True): 

47 found_web_document_response: WebDocumentResponse = WebDocumentResponse( 

48 id=found_document.id, 

49 name=found_document.name, 

50 description=found_document.description, 

51 document_type_id=found_document.document_type_id, 

52 account_id=found_document.account_id, 

53 web_url=found_web_document.web_url, 

54 web_url_hash=found_web_document.web_url_hash 

55 ) 

56 found_web_document_responses.append(found_web_document_response) 

57 

58 return found_web_document_responses 

59 

60 async def find_one_by_id_with_authorization(self, state: State, id: UUID) -> WebDocumentResponse: 

61 found_document: Document = await self.document_management.find_one_by_id_with_authorization( 

62 state=state, 

63 id=id 

64 ) 

65 found_web_document: WebDocument = await self.web_document_repository.find_one_by_id_and_account_id( 

66 session=state.session, 

67 id=id, 

68 account_id=state.authorized_session.account_id 

69 ) 

70 found_web_document_response: WebDocumentResponse = WebDocumentResponse( 

71 id=found_document.id, 

72 name=found_document.name, 

73 description=found_document.description, 

74 document_type_id=found_document.document_type_id, 

75 account_id=found_document.account_id, 

76 web_url=found_web_document.web_url, 

77 web_url_hash=found_web_document.web_url_hash 

78 ) 

79 return found_web_document_response 

80 

81 async def create_one(self, state: State, body: CreateOneBody) -> WebDocumentResponse: 

82 document_creator: Document = Document( 

83 id=uuid.uuid4(), 

84 name=body.name, 

85 description=body.description, 

86 document_type_id=DocumentTypeConstant.WEB, 

87 account_id=body.account_id 

88 ) 

89 created_document: Document = self.document_management.create_one_raw( 

90 state=state, 

91 document_creator=document_creator 

92 ) 

93 web_document_creator: WebDocument = WebDocument( 

94 id=created_document.id, 

95 web_url=body.web_url, 

96 web_url_hash=hashlib.sha256(body.web_url.encode()).hexdigest() 

97 ) 

98 created_web_document: WebDocument = self.create_one_raw( 

99 state=state, 

100 web_document_creator=web_document_creator 

101 ) 

102 web_document_response: WebDocumentResponse = WebDocumentResponse( 

103 id=created_document.id, 

104 name=created_document.name, 

105 description=created_document.description, 

106 document_type_id=created_document.document_type_id, 

107 account_id=created_document.account_id, 

108 web_url=created_web_document.web_url, 

109 web_url_hash=created_web_document.web_url_hash 

110 ) 

111 return web_document_response 

112 

113 def create_one_raw(self, state: State, web_document_creator: WebDocument) -> WebDocument: 

114 created_web_document: WebDocument = self.web_document_repository.create_one( 

115 session=state.session, 

116 web_document_creator=web_document_creator 

117 ) 

118 return created_web_document 

119 

120 async def patch_one_by_id_with_authorization(self, state: State, id: UUID, 

121 body: PatchOneBody) -> WebDocumentResponse: 

122 document_patcher: Document = Document( 

123 id=id, 

124 name=body.name, 

125 description=body.description, 

126 document_type_id=DocumentTypeConstant.WEB, 

127 account_id=body.account_id 

128 ) 

129 patched_document: Document = await self.document_management.patch_one_by_id_raw_with_authorization( 

130 state=state, 

131 id=id, 

132 document_patcher=document_patcher 

133 ) 

134 web_document_patcher: WebDocument = WebDocument( 

135 id=id, 

136 web_url=body.web_url, 

137 web_url_hash=hashlib.sha256(body.web_url.encode()).hexdigest() 

138 ) 

139 patched_web_document: WebDocument = await self.patch_one_by_id_raw_with_authorization( 

140 state=state, 

141 id=id, 

142 web_document_patcher=web_document_patcher 

143 ) 

144 patched_web_document_response: WebDocumentResponse = WebDocumentResponse( 

145 id=patched_document.id, 

146 name=patched_document.name, 

147 description=patched_document.description, 

148 document_type_id=patched_document.document_type_id, 

149 account_id=patched_document.account_id, 

150 web_url=patched_web_document.web_url, 

151 web_url_hash=patched_web_document.web_url_hash 

152 ) 

153 return patched_web_document_response 

154 

155 async def patch_one_by_id_raw_with_authorization(self, state: State, id: UUID, 

156 web_document_patcher: WebDocument) -> WebDocument: 

157 patched_web_document: WebDocument = await self.web_document_repository.patch_one_by_id_and_account_id( 

158 session=state.session, 

159 id=id, 

160 account_id=state.authorized_session.account_id, 

161 web_document_patcher=web_document_patcher, 

162 ) 

163 return patched_web_document 

164 

165 async def delete_one_by_id_with_authorization(self, state: State, id: UUID) -> WebDocumentResponse: 

166 deleted_web_document: WebDocument = await self.web_document_repository.delete_one_by_id_and_account_id( 

167 session=state.session, 

168 id=id, 

169 account_id=state.authorized_session.account_id 

170 ) 

171 deleted_document: Document = await self.document_management.delete_one_by_id_with_authorization( 

172 state=state, 

173 id=id 

174 ) 

175 deleted_web_document_response: WebDocumentResponse = WebDocumentResponse( 

176 id=deleted_web_document.id, 

177 name=deleted_document.name, 

178 description=deleted_document.description, 

179 document_type_id=deleted_document.document_type_id, 

180 account_id=deleted_document.account_id, 

181 web_url=deleted_web_document.web_url, 

182 web_url_hash=deleted_web_document.web_url_hash 

183 ) 

184 return deleted_web_document_response