Coverage for api/handlers/resources/drivers.py: 45%

1import asyncio 

2from typing import Any 

3 

4from bson import ObjectId 

5from database.entity import update_etag_and_text_search 

6from database.integration_1c.savers import remove_from_1c, update_in_1c 

7from database.updater import update_resource_in_order 

8from exceptions import BadParameterHTTPError, NotFoundHTTPError 

9from handlers.authorization.check_role import has_role 

10from handlers.grabbers.resources import drivers_data_grabber 

11from handlers.resources import documents_creation, documents_on_update 

12from handlers.resources.common import assert_resource_manage 

13from handlers.resources.processor import process_final_entity 

14from mongodb import docs_col, drivers_col 

15from pymongo import ReturnDocument 

16from pymongo.errors import DuplicateKeyError 

17from sotrans_models.models.misc.document import DocumentDBModel 

18from sotrans_models.models.resources.drivers import ( 

19 DriverCreateModel, 

20 DriverDBModel, 

21 DriverUpdateModel, 

22) 

23from sotrans_models.models.responses import GenericGetListResponse 

24from sotrans_models.models.roles import SotransRole 

25from sotrans_models.models.users import SotransOIDCUserModel 

26from sotrans_models.utils.text_mappers import get_drivers_text_search 

27from utils.data_grabber import BaseGetListQueryParams, adjust_search_query 

28from utils.dt_utils import get_current_datetime 

29from utils.helper import EntityHash, get_org_oid 

30 

31 

32async def on_create_driver( 

33 creation_data: DriverCreateModel, executor: SotransOIDCUserModel 

34) -> DriverDBModel | dict[str, Any]: 

35 org_id = get_org_oid(executor) 

36 data = creation_data.model_dump(exclude_unset=True) 

37 if DriverDBModel.documents[0] in data: 

38 del data["documents"] 

39 db_model = DriverDBModel( 

40 organization_id=org_id, 

41 created_at=get_current_datetime(), 

42 is_active=True, 

43 **data, 

44 ) 

45 try: 

46 created = await drivers_col.create(db_model.model_dump()) 

47 except DuplicateKeyError: 

48 raise BadParameterHTTPError("паспорт уже зарегистрирован") 

49 updated = await documents_creation( 

50 creation_data, drivers_col, created, executor 

51 ) 

52 return await process_final_entity( 

53 created if updated is None else updated, 

54 drivers_col, 

55 DriverDBModel, 

56 get_drivers_text_search, 

57 ) 

58 

59 

60async def on_get_drivers( 

61 params: BaseGetListQueryParams, 

62) -> GenericGetListResponse[DriverDBModel]: 

63 return await drivers_data_grabber.get_list(params) 

64 

65 

66async def on_get_driver( 

67 driver_id: ObjectId, executor: SotransOIDCUserModel 

68) -> DriverDBModel: 

69 return await assert_resource_manage(driver_id, drivers_col, executor) 

70 

71 

72async def on_update_driver( 

73 driver_id: ObjectId, 

74 update_data: DriverUpdateModel, 

75 executor: SotransOIDCUserModel, 

76) -> DriverDBModel: 

77 await assert_resource_manage( 

78 driver_id, drivers_col, executor, update_data.etag 

79 ) 

80 data = await documents_on_update( 

81 drivers_col, driver_id, update_data, executor 

82 ) 

83 etag_q = {"etag": update_data.etag} if update_data.etag is not None else {} 

84 try: 

85 driver = await drivers_col.collection.find_one_and_update( 

86 {"_id": driver_id} | etag_q, 

87 {"$set": data}, 

88 return_document=ReturnDocument.AFTER, 

89 ) 

90 except DuplicateKeyError: 

91 raise BadParameterHTTPError( 

92 "паспорт уже зарегистрирован в организации" 

93 ) 

94 if driver is None: 

95 raise NotFoundHTTPError("driver") 

96 await update_etag_and_text_search( 

97 driver, drivers_col, DriverDBModel, get_drivers_text_search 

98 ) 

99 dm = DriverDBModel(**driver) 

100 asyncio.create_task( 

101 update_in_1c(driver_id, drivers_col.collection_name, dm) 

102 ) 

103 await update_resource_in_order(driver_id, driver, "driver") 

104 return dm 

105 

106 

107async def on_delete_driver( 

108 driver_id: ObjectId, 

109 executor: SotransOIDCUserModel, 

110 etag: EntityHash | None, 

111): 

112 await assert_resource_manage(driver_id, drivers_col, executor, etag) 

113 await drivers_col.delete_by_id(driver_id) 

114 await docs_col.collection.update_many( 

115 {DocumentDBModel.object_id: driver_id}, 

116 {"$set": {DocumentDBModel.deleted_at: get_current_datetime()}}, 

117 ) 

118 await update_resource_in_order(driver_id, None, "driver") 

119 asyncio.create_task(remove_from_1c(driver_id, drivers_col.collection_name))