Coverage for api/handlers/orders/active_orders.py: 12%

1import asyncio 

2from typing import Any 

3 

4from bson import ObjectId 

5from database.orders import move_order_and_return 

6from exceptions import ( 

7 BadParameterHTTPError, 

8 NoAccessHTTPError, 

9 NotFoundHTTPError, 

10) 

11from handlers.authorization.check_role import has_role 

12from handlers.authorization.confidential import hide_model_fields_from_carrier 

13from handlers.grabbers.orders import orders_data_grabber 

14from mongodb import archive_col, orders_col 

15from pymongo import ReturnDocument 

16from services.notifications.director import notification_api 

17from sotrans_models.models.orders.order import ( 

18 OrderDBModel, 

19 OrderStageModel, 

20 OrderStageSubmitModel, 

21 OrderStageSuggestModel, 

22 OrderStatus, 

23 StageType, 

24 StopType, 

25) 

26from sotrans_models.models.responses import GenericGetListResponse 

27from sotrans_models.models.roles import SotransRole 

28from sotrans_models.models.users import SotransOIDCUserModel 

29from utils.data_grabber import ( 

30 BaseGetListQueryParams, 

31 adjust_search_query, 

32 update_search_query, 

33) 

34from utils.dt_utils import get_current_datetime 

35from utils.helper import get_org_oid 

36 

37 

38async def on_get_actives( 

39 user: SotransOIDCUserModel, params: BaseGetListQueryParams 

40) -> GenericGetListResponse[OrderDBModel]: 

41 params.where = adjust_search_query( 

42 params.where, OrderDBModel.status, {OrderStatus.active.value} 

43 ) 

44 if has_role(user, SotransRole.carrier_logistician) and not has_role( 

45 user, SotransRole.company_logistician 

46 ): 

47 carriers_query = { 

48 OrderDBModel.carrier.id: user.organization_id, # converted on json unpacking 

49 } 

50 params.where = update_search_query(params.where, carriers_query) 

51 active_orders = await orders_data_grabber.get_list(params, user) 

52 if not has_role(user, SotransRole.company_logistician): 

53 for order in active_orders.items: 

54 hide_model_fields_from_carrier(order) 

55 return active_orders 

56 

57 

58async def on_update_current_stage( 

59 order_id: ObjectId, 

60 stage: OrderStageSubmitModel, 

61 user: SotransOIDCUserModel, 

62) -> dict[str, Any] | OrderDBModel: 

63 restriction_q: dict = {OrderDBModel.status: OrderStatus.active.value} 

64 order = await orders_col.collection.find_one( 

65 {"id": order_id} | restriction_q 

66 ) 

67 if not order: 

68 raise NotFoundHTTPError("order") 

69 order_model = OrderDBModel(**order) 

70 if stage.id > len(order_model.stops or []) + 2: 

71 raise BadParameterHTTPError("stage_id") 

72 marked_stages = {s.id for s in order_model.completed_stages or []} 

73 if stage.id in marked_stages: 

74 raise BadParameterHTTPError("уже отмечено") 

75 if ( 

76 not has_role(user, SotransRole.drivers_bot) 

77 and not has_role(user, SotransRole.company_logistician) 

78 and order_model.carrier.id != get_org_oid(user) 

79 ): 

80 raise NoAccessHTTPError("order") 

81 i = 0 

82 ts = get_current_datetime() 

83 if order_model.stops: 

84 for i, s in enumerate(order_model.stops, start=1): 

85 if i == stage.id: 

86 stage = OrderStageModel( 

87 id=i, type=StageType.finished_stop, stop=s, timestamp=ts 

88 ) 

89 break 

90 if stage.id == i + 1: 

91 if order_model.completed_stages is None: 

92 order_model.completed_stages = [] 

93 for k, st in enumerate(order_model.stops, start=1): 

94 if k in marked_stages: 

95 continue 

96 order_model.completed_stages.append( 

97 OrderStageModel( 

98 id=k, type=StageType.finished_stop, stop=st, timestamp=ts 

99 ) 

100 ) 

101 stage = OrderStageModel( 

102 id=i + 1, type=StageType.ready_for_check, timestamp=ts 

103 ) 

104 elif stage.id == i + 2: 

105 if not has_role(user, SotransRole.company_logistician): 

106 raise NoAccessHTTPError("stage") 

107 stage = OrderStageModel(id=i + 2, type=StageType.ready, timestamp=ts) 

108 order_model.current_stage = stage 

109 if order_model.completed_stages: 

110 order_model.completed_stages.append(stage) 

111 else: 

112 order_model.completed_stages = [stage] 

113 

114 order_data = order_model.model_dump() 

115 

116 if stage.id == i + 2: 

117 order = await move_order_and_return( 

118 order_id, 

119 orders_col, 

120 archive_col, 

121 OrderStatus.archived, 

122 restriction_q, 

123 order_data=order_data, 

124 ) 

125 om = OrderDBModel(**order) 

126 asyncio.create_task(notification_api.order_archived(om)) 

127 return om 

128 update_data = { 

129 OrderDBModel.current_stage: order_data[OrderDBModel.current_stage], 

130 "completed_stages": order_data["completed_stages"], 

131 } 

132 updated = await orders_col.collection.find_one_and_update( 

133 {"id": order_id} | restriction_q, 

134 {"$set": update_data}, 

135 return_document=ReturnDocument.AFTER, 

136 ) 

137 if not updated: 

138 raise NotFoundHTTPError("order") 

139 if stage.id == i + 1: 

140 notification_api.company_order_completed(OrderDBModel(**updated)) 

141 return updated 

142 

143 

144async def on_get_stages_suggestions( 

145 order_id: ObjectId, 

146 user: SotransOIDCUserModel, 

147) -> list[OrderStageSuggestModel]: 

148 order = await orders_col.collection.find_one( 

149 {"id": order_id, OrderDBModel.status: OrderStatus.active.value} 

150 ) 

151 if not order: 

152 raise NotFoundHTTPError("order") 

153 order_model = OrderDBModel(**order) 

154 if ( 

155 not has_role(user, SotransRole.company_logistician) 

156 and order_model.current_stage 

157 and order_model.current_stage.type == StageType.ready_for_check 

158 ): 

159 return [] 

160 stages: list[OrderStageSuggestModel] = [] 

161 i = 0 

162 if order_model.stops: 

163 for i, s in enumerate(order_model.stops, start=1): 

164 stage = OrderStageModel(id=i, type=StageType.finished_stop, stop=s) 

165 found = False 

166 if order_model.completed_stages: 

167 for completed_stage in order_model.completed_stages: 

168 if ( 

169 completed_stage.id == stage.id 

170 and completed_stage.stop == stage.stop 

171 ): 

172 found = True 

173 if found: 

174 continue 

175 stop_type_for_user = ( 

176 "Погрузка" if s.stop_type == StopType.loading else "Разгрузка" 

177 ) 

178 geo = s.address.city or s.address.line 

179 stages.append( 

180 OrderStageSuggestModel( 

181 id=i, label=f"{stop_type_for_user} {geo} пройдена" 

182 ) 

183 ) 

184 i += 1 

185 stage = OrderStageModel(id=i, type=StageType.ready_for_check) 

186 if has_role(user, SotransRole.company_logistician): 

187 stages.append(OrderStageSuggestModel(id=i + 1, label="Готов")) 

188 if order_model.completed_stages: 

189 for completed_stage in order_model.completed_stages: 

190 if completed_stage.id == stage.id: 

191 break 

192 else: 

193 stages.append( 

194 OrderStageSuggestModel(id=i, label="Готов к проверке") 

195 ) 

196 else: 

197 stages.append(OrderStageSuggestModel(id=i, label="Готов к проверке")) 

198 return stages