Coverage for api/handlers/misc/documents.py: 16%
153 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1import datetime
2from typing import Any, Callable
4from bson import ObjectId
5from config import EMBEDDED_DOCUMENTS_LIMIT
6from database.entity import update_etag_and_text_search
7from database.integration_1c.entity_checker import (
8 check_resource_movable_to_updates,
9)
10from database.integration_1c.savers import save_entity_to_updates
11from dbcc import MongoTableEngine
12from exceptions import (
13 BadParameterHTTPError,
14 NoAccessHTTPError,
15 NotAcceptableHTTPError,
16 NotFoundHTTPError,
17)
18from handlers.authorization.check_role import has_role
19from handlers.grabbers.documents import docs_grabber
20from mongodb import (
21 db,
22 docs_col,
23 drivers_col,
24 orders_col,
25 orgs_col,
26 trailers_col,
27 trucks_col,
28)
29from pymongo import ReturnDocument
30from pymongo.errors import WriteError
31from sotrans_models.models._mongo import PydanticObjectId
32from sotrans_models.models.misc.document import (
33 DocumentCreateModel,
34 DocumentDBModel,
35 DocumentTypeModel,
36 DocumentTypingAlias,
37 DocumentUpdateModel,
38 DriverDocumentType,
39 OrganizationDocumentType,
40 RequestDocumentType,
41 TrailerDocumentType,
42 TruckDocumentType,
43 UserDocumentType,
44)
45from sotrans_models.models.resources.trucks import TruckDBModel
46from sotrans_models.models.responses import GenericGetListResponse
47from sotrans_models.models.roles import SotransRole
48from sotrans_models.models.users import SotransOIDCUserModel
49from sotrans_models.utils.text_mappers import get_documents_text_search
50from utils.data_grabber import (
51 BaseGetListQueryParams,
52 BaseGetOneQueryParams,
53 update_search_query,
54)
55from utils.dt_utils import get_current_datetime
56from utils.helper import EntityHash, etag_detalizer, get_hash, get_org_oid
59def docs_restriction_query_builder(executor: SotransOIDCUserModel):
60 restriction_query = {DocumentDBModel.deleted_at: None}
61 if not has_role(executor, SotransRole.company_logistician):
62 restriction_query.update(
63 {DocumentDBModel.organization_id: get_org_oid(executor)}
64 )
65 if not has_role(executor, SotransRole.carrier_director):
66 restriction_query.update(
67 {
68 DocumentDBModel.type: {
69 "$nin": [t.value for t in OrganizationDocumentType]
70 }
71 }
72 )
73 return restriction_query
76async def type_document_strategy(
77 document_type: DocumentTypingAlias, coro: Callable
78):
79 if isinstance(document_type, (UserDocumentType, RequestDocumentType)):
80 await coro(orders_col)
81 elif isinstance(document_type, TruckDocumentType):
82 await coro(trucks_col)
83 elif isinstance(document_type, TrailerDocumentType):
84 await coro(trailers_col)
85 elif isinstance(document_type, DriverDocumentType):
86 await coro(drivers_col)
89def check_valid_until(
90 document: DocumentCreateModel, current_dt: datetime.datetime
91):
92 if document.valid_until:
93 naive_current = current_dt.replace(tzinfo=None)
94 valid_until_comparable = document.valid_until.replace(tzinfo=None)
95 if document.valid_until and valid_until_comparable <= naive_current:
96 raise BadParameterHTTPError("Документ просрочен")
99async def get_related_with_checks(
100 object_id: ObjectId, col_name: str
101) -> dict[str, Any]:
102 related = await db[col_name].find_single(
103 "_id",
104 object_id,
105 )
106 if related is None:
107 raise NotFoundHTTPError("сущность под object_id")
108 return related
111def try_fill_org_id(
112 related: dict, document_data: dict[str, Any], user: SotransOIDCUserModel
113):
114 org_oid = related.get(TruckDBModel.organization_id)
115 if org_oid is not None:
116 if not has_role(
117 user, SotransRole.company_logistician
118 ) and user.organization_id != str(org_oid):
119 raise NoAccessHTTPError("другая организация")
120 document_data["organization_id"] = org_oid
123async def on_create_document(
124 document: DocumentCreateModel, user: SotransOIDCUserModel
125):
126 if document.collection and document.collection not in (
127 trucks_col.collection_name,
128 trailers_col.collection_name,
129 drivers_col.collection_name,
130 orders_col.collection_name,
131 orgs_col.collection_name,
132 docs_col.collection_name,
133 ):
134 raise BadParameterHTTPError("несуществующая коллекция")
135 current_time = get_current_datetime()
136 check_valid_until(document, current_time)
138 document_data = document.model_dump()
139 if not document.collection:
140 if document.object_id:
141 if isinstance(document.type, TruckDocumentType):
142 col = trucks_col.collection_name
143 elif isinstance(document.type, TrailerDocumentType):
144 col = trailers_col.collection_name
145 elif isinstance(document.type, DriverDocumentType):
146 col = drivers_col.collection_name
147 elif isinstance(document.type, UserDocumentType):
148 col = orders_col.collection_name
149 else:
150 col = orgs_col.collection_name
151 else:
152 col = docs_col.collection_name
153 else:
154 col = document.collection
156 if document.object_id:
157 if (
158 not has_role(user, SotransRole.company_logistician)
159 and col == orgs_col.collection_name
160 ):
161 org_oid = get_org_oid(user)
162 if org_oid != document.object_id:
163 raise NoAccessHTTPError("организация")
164 related = await get_related_with_checks(document.object_id, col)
165 if col == orgs_col.collection_name:
166 document_data["organization_id"] = document.object_id
167 else:
168 try_fill_org_id(related, document_data, user)
169 document_data["verification"] = None
170 document_data["created_at"] = current_time
171 document_data["collection"] = col
172 etag = get_hash(document_data)
173 inserted_main_doc = await docs_col.create(
174 {
175 **document_data,
176 **{
177 "etag": etag,
178 DocumentDBModel.text_search: get_documents_text_search(
179 DocumentDBModel(**document_data)
180 ),
181 },
182 }
183 )
185 async def update_or_create_documents(collection: MongoTableEngine):
186 try:
187 await collection.collection.update_one(
188 {
189 "_id": document.object_id,
190 },
191 {"$push": {TruckDBModel.documents[0]: inserted_main_doc}},
192 )
193 except (
194 WriteError,
195 AttributeError,
196 ): # MongoMock throws AttributeError
197 await collection.update_by_id(
198 document.object_id, {"documents": [inserted_main_doc]}
199 )
201 if document.object_id:
202 if (docs := related.get("documents")) is None or len(
203 docs
204 ) < EMBEDDED_DOCUMENTS_LIMIT:
205 await update_or_create_documents(db[col])
206 if col in (
207 trucks_col.collection_name,
208 trailers_col.collection_name,
209 drivers_col.collection_name,
210 ) and await check_resource_movable_to_updates(
211 related, col, document.type
212 ):
213 # this is impossible as check_resource_movable_to_updates needs 1 doc at least
214 # if related.get("documents") is None:
215 # related["documents"] = [inserted_main_doc]
216 # else:
217 related["documents"].append(inserted_main_doc)
218 await save_entity_to_updates(col, related)
220 return inserted_main_doc
223async def on_delete_document(
224 executor: SotransOIDCUserModel, doc_id: ObjectId, etag: EntityHash | None
225):
226 restriction_query = docs_restriction_query_builder(executor)
227 etag_q = {"etag": etag} if etag is not None else {}
228 document = await docs_col.collection.find_one_and_update(
229 {"_id": doc_id} | restriction_query | etag_q,
230 {"$set": {DocumentDBModel.deleted_at: datetime.datetime.utcnow()}},
231 )
232 if document is None:
233 await etag_detalizer(
234 docs_col, etag_q, {"_id": doc_id} | restriction_query
235 )
236 raise NotFoundHTTPError("документ")
237 doc_type = DocumentTypeModel(document_type=document["type"])
239 async def remove_from_collection(collection: MongoTableEngine):
240 await collection.collection.update_one(
241 {TruckDBModel.documents[0].id: doc_id},
242 {"$pull": {TruckDBModel.documents[0]: {"_id": doc_id}}},
243 )
245 await type_document_strategy(
246 doc_type.document_type, remove_from_collection
247 )
250async def on_get_documents(
251 executor: SotransOIDCUserModel, params: BaseGetListQueryParams
252):
253 documents_filter: dict = {DocumentDBModel.deleted_at: None}
254 if not (
255 has_role(executor, SotransRole.carrier_director)
256 or has_role(executor, SotransRole.company_logistician)
257 ):
258 db_doc_types = [t.value for t in OrganizationDocumentType]
259 documents_filter.update({"type": {"$nin": db_doc_types}})
260 if not has_role(executor, SotransRole.company_logistician):
261 org_oid = get_org_oid(executor)
262 organization = await orgs_col.collection.find_one(
263 {"_id": org_oid},
264 projection={"_id": 1},
265 )
266 documents_filter.update({"organization_id": organization["_id"]})
267 params.where = update_search_query(params.where, documents_filter)
268 return await docs_grabber.get_list(params, executor)
271async def on_get_documents_new(
272 executor: SotransOIDCUserModel,
273 collection: str,
274 object_id: PydanticObjectId,
275 params: BaseGetListQueryParams,
276) -> GenericGetListResponse[DocumentDBModel]:
277 if collection and collection not in (
278 trucks_col.collection_name,
279 trailers_col.collection_name,
280 drivers_col.collection_name,
281 orders_col.collection_name,
282 orgs_col.collection_name,
283 docs_col.collection_name,
284 ):
285 raise BadParameterHTTPError("несуществующая коллекция")
286 if not has_role(executor, SotransRole.company_logistician):
287 params.where = update_search_query(
288 params.where,
289 # types are correct as serialized!
290 {DocumentDBModel.organization_id: executor.organization_id},
291 )
292 params.where = update_search_query(
293 params.where, {DocumentDBModel.object_id: str(object_id)}
294 )
295 params.where = update_search_query(
296 params.where, {DocumentDBModel.collection: collection}
297 )
298 params.where = update_search_query(
299 params.where, {DocumentDBModel.deleted_at: None}
300 )
301 return await docs_grabber.get_list(params)
304async def on_get_document(
305 executor: SotransOIDCUserModel,
306 doc_id: ObjectId,
307 params: BaseGetOneQueryParams,
308) -> DocumentDBModel | dict[str, Any]:
309 restriction_query = docs_restriction_query_builder(executor)
310 return await docs_grabber.get_one_by_id_with_pattern(
311 doc_id, params, restriction_query
312 )
315async def on_update_document(
316 executor: SotransOIDCUserModel, doc_id: ObjectId, doc: DocumentUpdateModel
317) -> DocumentDBModel:
318 if doc.object_id:
319 raise NotAcceptableHTTPError("id объекта не подлежит изменению")
320 restriction_query = docs_restriction_query_builder(executor)
321 up_payload = doc.model_dump(exclude_none=True)
322 etag_q = {"etag": doc.etag} if doc.etag else {}
323 document = await docs_col.collection.find_one_and_update(
324 {"_id": doc_id} | restriction_query | etag_q,
325 {"$set": up_payload},
326 return_document=ReturnDocument.AFTER,
327 )
328 if document is None:
329 await etag_detalizer(
330 docs_col, etag_q, {"_id": doc_id} | restriction_query
331 )
332 raise NotFoundHTTPError("документ")
333 await update_etag_and_text_search(
334 document, docs_col, DocumentDBModel, get_documents_text_search
335 )
336 doc_type = DocumentTypeModel(document_type=document["type"])
338 async def update_by_type(collection: MongoTableEngine):
339 array_update = {f"documents.$.{k}": up_payload[k] for k in up_payload}
340 await collection.collection.update_one(
341 {TruckDBModel.documents[0].id: doc_id}, {"$set": array_update}
342 )
344 await type_document_strategy(doc_type.document_type, update_by_type)
346 return document