Coverage for api/services/recommendations.py: 21%

1from typing import Any 

2 

3import aiohttp 

4import config 

5from bson import ObjectId 

6from exceptions import StopsLimitedInfo 

7from services.microservice_connector import RecommendationsConnector 

8from sotrans_models.models.orders.order import ( 

9 OrderDBModel, 

10 OrderUpdateModel, 

11 RecommendationMatch, 

12 RecommendationServiceOrderModel, 

13 RecommendationServiceUpdateOrderModel, 

14 StopModel, 

15) 

16 

17 

18def check_location(stop: StopModel) -> bool: 

19 if not stop.address.location: 

20 return False 

21 if not ( 

22 stop.address.location.latitude and stop.address.location.longitude 

23 ): 

24 return False 

25 return True 

26 

27 

28def normalize_stops(order: OrderDBModel | OrderUpdateModel): 

29 order.stops = list(filter(lambda stop: stop.datetime, order.stops)) 

30 if len(order.stops) < 2: 

31 raise StopsLimitedInfo 

32 order.stops.sort(key=lambda stop: stop.datetime) 

33 if not ( 

34 check_location(order.stops[0]) and check_location(order.stops[-1]) 

35 ): 

36 raise StopsLimitedInfo 

37 

38 

39def normalize_datetime_to_msc(order_data: dict[str, Any]): 

40 for date in ("finish_date", "start_date"): 

41 date_value: str | None = order_data.get(date) 

42 if date_value: 

43 date_value = date_value.rstrip("Z") 

44 if len(date_value) > 5 and date_value[-6] not in ("-", "+"): 

45 order_data[date] = f"{date_value}+03:00" 

46 

47 

48class APIQuerier: 

49 API_URL = config.RecommendationAPIConfig.RECOMMENDATION_API_URL 

50 

51 def __init__(self, connector: RecommendationsConnector): 

52 self.connector = connector 

53 

54 def _form_rec_domain( 

55 self, order_resource: OrderDBModel, *, update: bool 

56 ) -> dict[str, Any]: 

57 model = ( 

58 RecommendationServiceUpdateOrderModel 

59 if update 

60 else RecommendationServiceOrderModel 

61 ) 

62 create_entity = model( 

63 id=order_resource.id, 

64 start_date=order_resource.stops[0].datetime, 

65 finish_date=order_resource.stops[-1].datetime, 

66 start_location=order_resource.stops[0].address.location, 

67 finish_location=order_resource.stops[-1].address.location, 

68 volume=order_resource.truck_body.volume, 

69 weight=order_resource.truck_body.weight, 

70 body=order_resource.truck_body.body_type, 

71 loading=order_resource.truck_body.loading_type, 

72 ) 

73 json = create_entity.model_dump(mode="json") 

74 normalize_datetime_to_msc(json) 

75 return json 

76 

77 async def create_target_and_recommendation_order( 

78 self, order: OrderDBModel, *, target: bool = True 

79 ): 

80 if not order.stops: 

81 return 

82 if not order.truck_body.loading_type: 

83 return 

84 

85 order_resource = order.model_copy(deep=True) 

86 try: 

87 normalize_stops(order_resource) 

88 except StopsLimitedInfo: 

89 return 

90 

91 create_entity = self._form_rec_domain(order_resource, update=False) 

92 await self.connector.make_request( 

93 self.API_URL, 

94 "/target-orders" if target else "/recommendation-orders", 

95 "POST", 

96 json=create_entity, 

97 ) 

98 

99 async def update_order( 

100 self, oid: ObjectId, order: OrderDBModel, *, target: bool 

101 ): 

102 order_copy = order.model_copy(deep=True) 

103 if order.stops: 

104 try: 

105 normalize_stops(order_copy) 

106 except StopsLimitedInfo: 

107 return 

108 data = self._form_rec_domain(order_copy, update=True) 

109 api_req_url = ( 

110 f"/target-orders/{oid}" 

111 if target 

112 else f"/recommendation-orders/{oid}" 

113 ) 

114 await self.connector.make_request( 

115 self.API_URL, api_req_url, "PATCH", json=data 

116 ) 

117 

118 async def remove_order(self, oid: ObjectId, *, target: bool): 

119 api_del_url = ( 

120 f"/target-orders/{oid}" 

121 if target 

122 else f"/recommendation-orders/{oid}" 

123 ) 

124 await self.connector.make_request(self.API_URL, api_del_url, "DELETE") 

125 

126 async def get_recommendations_for_target( 

127 self, oid: ObjectId 

128 ) -> list[RecommendationMatch]: 

129 api_get_url = f"/target-orders/{oid}/recommendations" 

130 try: 

131 recommendations = await self.connector.make_request( 

132 self.API_URL, api_get_url, "GET" 

133 ) 

134 except aiohttp.ClientError: 

135 return [] 

136 if not recommendations: 

137 return [] 

138 return [ 

139 RecommendationMatch(**recommendation) 

140 for recommendation in recommendations 

141 ] 

142 

143 

144suggestion_api = APIQuerier(RecommendationsConnector())