Coverage for apps/inners/use_cases/managements/file_document_management.py: 67%

1import hashlib 

2import uuid 

3from typing import List, Optional, Dict, Any 

4from uuid import UUID 

5 

6from starlette.datastructures import State 

7 

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

9from apps.inners.models.daos.file_document import FileDocument 

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

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

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

13from apps.inners.models.dtos.contracts.responses.managements.documents.file_document_response import \ 

14 FileDocumentResponse 

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

16from apps.outers.repositories.file_document_repository import FileDocumentRepository 

17 

18 

19class FileDocumentManagement: 

20 def __init__( 

21 self, 

22 document_management: DocumentManagement, 

23 file_document_repository: FileDocumentRepository, 

24 ): 

25 self.document_management: DocumentManagement = document_management 

26 self.file_document_repository: FileDocumentRepository = file_document_repository 

27 

28 def get_file_metadata(self, file_document: FileDocument) -> Dict[str, Any]: 

29 file_url: str = self.file_document_repository.get_object_url( 

30 object_name=file_document.file_name 

31 ) 

32 file_metadata: Dict[str, Any] = { 

33 "file_url": file_url 

34 } 

35 return file_metadata 

36 

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

38 List[ 

39 FileDocumentResponse]: 

40 found_file_documents: List[ 

41 FileDocument 

42 ] = await self.file_document_repository.find_many_by_account_id_with_pagination( 

43 session=state.session, 

44 account_id=state.authorized_session.account_id, 

45 page_position=page_position, 

46 page_size=page_size 

47 ) 

48 found_documents: List[ 

49 Document 

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

51 session=state.session, 

52 ids=[found_file_document.id for found_file_document in found_file_documents], 

53 account_id=state.authorized_session.account_id 

54 ) 

55 found_file_document_responses: List[FileDocumentResponse] = [] 

56 for found_document, found_file_document in zip(found_documents, found_file_documents, strict=True): 

57 found_file_document_response: FileDocumentResponse = FileDocumentResponse( 

58 id=found_document.id, 

59 name=found_document.name, 

60 description=found_document.description, 

61 document_type_id=found_document.document_type_id, 

62 account_id=found_document.account_id, 

63 file_name=found_file_document.file_name, 

64 file_data_hash=found_file_document.file_data_hash, 

65 file_metadata=self.get_file_metadata(found_file_document) 

66 ) 

67 found_file_document_responses.append(found_file_document_response) 

68 

69 return found_file_document_responses 

70 

71 async def find_one_by_id_with_authorization(self, state: State, id: UUID) -> FileDocumentResponse: 

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

73 state=state, 

74 id=id 

75 ) 

76 found_file_document: FileDocument = await self.file_document_repository.find_one_by_id_and_account_id( 

77 session=state.session, 

78 id=id, 

79 account_id=state.authorized_session.account_id 

80 ) 

81 found_file_document_response: FileDocumentResponse = FileDocumentResponse( 

82 id=found_document.id, 

83 name=found_document.name, 

84 description=found_document.description, 

85 document_type_id=found_document.document_type_id, 

86 account_id=found_document.account_id, 

87 file_name=found_file_document.file_name, 

88 file_data_hash=found_file_document.file_data_hash, 

89 file_metadata=self.get_file_metadata(found_file_document) 

90 ) 

91 return found_file_document_response 

92 

93 async def create_one(self, state: State, body: CreateOneBody) -> FileDocumentResponse: 

94 document_creator: Document = Document( 

95 id=uuid.uuid4(), 

96 name=body.name, 

97 description=body.description, 

98 document_type_id=DocumentTypeConstant.FILE, 

99 account_id=body.account_id 

100 ) 

101 created_document: Document = self.document_management.create_one_raw( 

102 state=state, 

103 document_creator=document_creator 

104 ) 

105 file_data: bytes = await body.file_data.read() 

106 await body.file_data.close() 

107 file_document_creator: FileDocument = FileDocument( 

108 id=created_document.id, 

109 file_name=f"{uuid.uuid4()}_{body.file_name}", 

110 file_data_hash=hashlib.sha256(file_data).hexdigest() 

111 ) 

112 created_file_document: FileDocument = self.create_one_raw( 

113 state=state, 

114 file_document_creator=file_document_creator, 

115 file_data=file_data 

116 ) 

117 file_document_response: FileDocumentResponse = FileDocumentResponse( 

118 id=created_document.id, 

119 name=created_document.name, 

120 description=created_document.description, 

121 document_type_id=created_document.document_type_id, 

122 account_id=created_document.account_id, 

123 file_name=created_file_document.file_name, 

124 file_data_hash=created_file_document.file_data_hash, 

125 file_metadata=self.get_file_metadata(created_file_document) 

126 ) 

127 return file_document_response 

128 

129 def create_one_raw(self, state: State, file_document_creator: FileDocument, 

130 file_data: Optional[bytes] = None) -> FileDocument: 

131 created_file_document: FileDocument = self.file_document_repository.create_one( 

132 session=state.session, 

133 file_document_creator=file_document_creator, 

134 file_data=file_data 

135 ) 

136 return created_file_document 

137 

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

139 body: PatchOneBody) -> FileDocumentResponse: 

140 document_patcher: Document = Document( 

141 id=id, 

142 name=body.name, 

143 description=body.description, 

144 document_type_id=DocumentTypeConstant.FILE, 

145 account_id=body.account_id 

146 ) 

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

148 state=state, 

149 id=id, 

150 document_patcher=document_patcher 

151 ) 

152 found_file_document: FileDocument = await self.file_document_repository.find_one_by_id_and_account_id( 

153 session=state.session, 

154 id=id, 

155 account_id=state.authorized_session.account_id 

156 ) 

157 file_document_patcher: FileDocument = FileDocument() 

158 if body.file_name == found_file_document.file_name: 

159 file_document_patcher.file_name = body.file_name 

160 else: 

161 file_document_patcher.file_name = f"{uuid.uuid4()}_{body.file_name}" 

162 file_data: Optional[bytes] = None 

163 if body.file_data is not None: 

164 file_data = await body.file_data.read() 

165 await body.file_data.close() 

166 file_document_patcher.file_data_hash = hashlib.sha256(file_data).hexdigest() 

167 patched_file_document: FileDocument = await self.patch_one_by_id_raw_with_authorization( 

168 state=state, 

169 id=id, 

170 file_document_patcher=file_document_patcher, 

171 file_data=file_data 

172 ) 

173 patched_file_document_response: FileDocumentResponse = FileDocumentResponse( 

174 id=patched_document.id, 

175 name=patched_document.name, 

176 description=patched_document.description, 

177 document_type_id=patched_document.document_type_id, 

178 account_id=patched_document.account_id, 

179 file_name=patched_file_document.file_name, 

180 file_data_hash=patched_file_document.file_data_hash, 

181 file_metadata=self.get_file_metadata(patched_file_document) 

182 ) 

183 return patched_file_document_response 

184 

185 async def patch_one_by_id_raw_with_authorization(self, state: State, id: UUID, file_document_patcher: FileDocument, 

186 file_data: Optional[bytes] = None) -> FileDocument: 

187 patched_file_document: FileDocument = await self.file_document_repository.patch_one_by_id_and_account_id( 

188 session=state.session, 

189 id=id, 

190 account_id=state.authorized_session.account_id, 

191 file_document_patcher=file_document_patcher, 

192 file_data=file_data 

193 ) 

194 return patched_file_document 

195 

196 async def delete_one_by_id_with_authorization(self, state: State, id: UUID) -> FileDocumentResponse: 

197 deleted_file_document: FileDocument = await self.file_document_repository.delete_one_by_id_and_account_id( 

198 session=state.session, 

199 id=id, 

200 account_id=state.authorized_session.account_id 

201 ) 

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

203 state=state, 

204 id=id 

205 ) 

206 deleted_file_document_response: FileDocumentResponse = FileDocumentResponse( 

207 id=deleted_document.id, 

208 name=deleted_document.name, 

209 description=deleted_document.description, 

210 document_type_id=deleted_document.document_type_id, 

211 account_id=deleted_document.account_id, 

212 file_name=deleted_file_document.file_name, 

213 file_data_hash=deleted_file_document.file_data_hash, 

214 file_metadata=None 

215 ) 

216 return deleted_file_document_response