Coverage for api/handlers/resources/__init__.py: 23%

1from typing import Any 

2 

3from bson import ObjectId 

4from database.text_search.mongo_search import update_query_by_search 

5from dbcc import MongoTableEngine 

6from exceptions import NoAccessHTTPError, NotFoundHTTPError 

7from handlers.authorization.check_role import has_role 

8from handlers.misc.documents import on_create_document 

9from sotrans_models.models.misc.verification import VerificationStatus 

10from sotrans_models.models.organizations import OrganizationCreateModel 

11from sotrans_models.models.resources.drivers import ( 

12 DriverCreateModel, 

13 DriverDBModel, 

14 DriverStatus, 

15 DriverUpdateModel, 

16) 

17from sotrans_models.models.resources.trailers import ( 

18 TrailerCreateModel, 

19 TrailerDBModel, 

20 TrailerUpdateModel, 

21) 

22from sotrans_models.models.resources.trucks import ( 

23 TruckCreateModel, 

24 TruckDBModel, 

25 TruckStatus, 

26 TruckUpdateModel, 

27) 

28from sotrans_models.models.roles import SotransRole 

29from sotrans_models.models.users import SotransOIDCUserModel 

30from utils.helper import get_org_oid 

31 

32 

33async def on_get_resources_for_org( 

34 collection: MongoTableEngine, 

35 limit: int, 

36 skip: int, 

37 user: SotransOIDCUserModel | None = None, 

38 search_q: str | None = None, 

39 org_id: ObjectId | None = None, 

40 for_suggestion: bool = False, 

41) -> list[TruckDBModel | TrailerDBModel | DriverDBModel]: 

42 if limit > 200: 

43 limit = 200 

44 if not has_role( 

45 user, SotransRole.company_logistician 

46 ) and org_id != get_org_oid(user): 

47 raise NoAccessHTTPError("организация") 

48 if not org_id: 

49 org_id = get_org_oid(user) 

50 query = {TruckDBModel.organization_id: org_id} 

51 if search_q: 

52 await update_query_by_search(collection, search_q, query) 

53 if for_suggestion is True: 

54 query |= { 

55 DriverDBModel.verification.status: VerificationStatus.accepted.value, 

56 DriverDBModel.status: { 

57 "$nin": [ 

58 DriverStatus.blocked.value, 

59 TruckStatus.under_repair.value, 

60 ] 

61 }, 

62 } 

63 resources = await collection.find_batch( 

64 query, skip, limit, [(DriverDBModel.verification.status, -1)] 

65 ) 

66 if collection.collection_name == "drivers": 

67 return [DriverDBModel(**d) for d in resources] 

68 if collection.collection_name == "trucks": 

69 return [TruckDBModel(**r) for r in resources] 

70 return [TrailerDBModel(**r) for r in resources] 

71 

72 

73async def documents_creation( 

74 creation_data: TrailerCreateModel 

75 | TruckCreateModel 

76 | DriverCreateModel 

77 | OrganizationCreateModel, 

78 collection: MongoTableEngine, 

79 created: dict[str, Any], 

80 executor: SotransOIDCUserModel, 

81) -> dict[str, Any] | None: 

82 if creation_data.documents: 

83 oid = created.get("_id") 

84 for doc in creation_data.documents: 

85 doc.collection = collection.collection_name 

86 doc.object_id = oid 

87 await on_create_document(doc, executor) 

88 updated = await collection.find_single("_id", oid) 

89 if updated is None: 

90 raise NotFoundHTTPError("ресурс") 

91 return updated 

92 return None 

93 

94 

95async def documents_on_update( 

96 collection: MongoTableEngine, 

97 oid: ObjectId, 

98 update_data: DriverUpdateModel | TrailerUpdateModel | TruckUpdateModel, 

99 executor: SotransOIDCUserModel, 

100) -> dict[str, Any]: 

101 data = update_data.model_dump(exclude_unset=True) 

102 if update_data.documents: 

103 del data["documents"] 

104 for doc in update_data.documents: 

105 doc.collection = collection.collection_name 

106 doc.object_id = oid 

107 await on_create_document(doc, executor) 

108 return data