Coverage for api/handlers/orders/bids.py: 17%

1import asyncio 

2import datetime 

3from typing import Any 

4 

5from bson import ObjectId 

6from exceptions import ( 

7 BadParameterHTTPError, 

8 NoAccessHTTPError, 

9 NotAcceptableHTTPError, 

10 NotFoundHTTPError, 

11) 

12from handlers.authorization.check_role import has_role 

13from handlers.grabbers.biding import bids_data_grabber 

14from handlers.grabbers.orders import orders_data_grabber 

15from mongodb import bids_col, orders_col, orgs_col, users_col 

16from services.notifications.director import notification_api 

17from sotrans_models.models.orders.bid import BidCreateModel, BidDBModel 

18from sotrans_models.models.orders.order import OrderDBModel 

19from sotrans_models.models.responses import GenericGetListResponse 

20from sotrans_models.models.roles import SotransRole 

21from sotrans_models.models.users import ( 

22 SotransOIDCUserModel, 

23 SotransUserDBFieldsModel, 

24) 

25from sotrans_models.utils.text_mappers import get_bids_text_search 

26from utils.data_grabber import BaseGetListQueryParams, MongoDataGrabber 

27from utils.dt_utils import get_current_datetime, get_datetime_tz_aware 

28from utils.helper import get_org_oid 

29 

30 

31async def _assert_bid_actions(bid_id: ObjectId, user: SotransOIDCUserModel): 

32 bid_data = await bids_col.find_single("_id", bid_id) 

33 if not bid_data: 

34 raise NotFoundHTTPError("bid") 

35 bid = BidDBModel(**bid_data) 

36 order_data = await orders_data_grabber.collection.find_single( 

37 "id", bid.order_id 

38 ) 

39 if not order_data: 

40 raise NotAcceptableHTTPError("Order moved.") 

41 order = OrderDBModel(**order_data) 

42 now = get_current_datetime() 

43 now = now.replace(tzinfo=None) 

44 if order.auction_end_time < now: 

45 raise NotAcceptableHTTPError("Auction is over") 

46 if user.sub != bid.owner.id: 

47 raise NoAccessHTTPError("bid") 

48 

49 

50def check_restrictions( 

51 order_data: dict[str, Any], 

52 bid: BidCreateModel, 

53 ts: datetime.datetime, 

54 lowest_bid: dict[str, Any], 

55): 

56 if bid.value <= 0: 

57 raise BadParameterHTTPError("значение") 

58 if not order_data: 

59 raise NotFoundHTTPError("заказ") 

60 order = OrderDBModel(**order_data) 

61 previous_bid_value = BidDBModel(**lowest_bid).value if lowest_bid else None 

62 previous_price = previous_bid_value or order.start_price 

63 if previous_price and (previous_price - bid.value) % order.rate_step != 0: 

64 raise BadParameterHTTPError("значение для шага и предыдущей ставки") 

65 if get_datetime_tz_aware(order.auction_end_time) <= ts: 

66 raise NotAcceptableHTTPError("аукцион завершён") 

67 if ( 

68 lowest_bid 

69 and (next_bid_val := previous_price - order.rate_step) < bid.value 

70 ): 

71 if next_bid_val > 0: 

72 raise NotAcceptableHTTPError(f"максимум: {next_bid_val}") 

73 else: 

74 raise NotAcceptableHTTPError("заказ занят") 

75 

76 

77async def on_make_bid( 

78 user: SotransOIDCUserModel, bid: BidCreateModel, order_id: ObjectId 

79) -> BidDBModel: 

80 order_data = await orders_data_grabber.collection.find_single( 

81 "id", order_id 

82 ) 

83 created_at = get_current_datetime() 

84 lowest_bid = await bids_col.collection.find_one( 

85 {BidDBModel.order_id: order_id}, sort=[("value", 1)] 

86 ) 

87 check_restrictions(order_data, bid, created_at, lowest_bid) 

88 bid_owner = await users_col.collection.find_one( 

89 {SotransUserDBFieldsModel.id: user.sub} 

90 ) 

91 bid_org_id = get_org_oid(user) 

92 bid_org = await orgs_col.find_single( 

93 "_id", bid_org_id, projection={"documents": 0} 

94 ) 

95 if bid_org is None: 

96 raise NotFoundHTTPError("пользователь не в организации") 

97 order = OrderDBModel(**order_data) 

98 bid_for_db = BidDBModel( 

99 value=bid.value, 

100 order_id=order_id, 

101 created_at=created_at, 

102 owner=bid_owner, 

103 client_id=order.client.id if order.client else None, 

104 carrier=bid_org, 

105 ) 

106 bid_for_db.text_search = get_bids_text_search(bid_for_db) 

107 bid_for_db = bid_for_db.model_dump() 

108 await bids_col.collection.delete_one( 

109 { 

110 BidDBModel.order_id: order_id, 

111 BidDBModel.carrier.id: bid_org["_id"], 

112 } 

113 ) 

114 created_bid = BidDBModel(**await bids_col.create(bid_for_db)) 

115 created_bid_data = created_bid.model_dump() 

116 await orders_col.collection.update_one( 

117 {"id": order.id}, {"$set": {OrderDBModel.best_bid: created_bid_data}} 

118 ) 

119 order_data[OrderDBModel.best_bid] = created_bid_data 

120 if lowest_bid: 

121 asyncio.create_task( 

122 notification_api.better_bid( 

123 OrderDBModel(**order_data), BidDBModel(**lowest_bid) 

124 ) 

125 ) 

126 return created_bid 

127 

128 

129async def on_bid_deletion(user: SotransOIDCUserModel, bid_id: ObjectId): 

130 await _assert_bid_actions(bid_id, user) 

131 bids_motor_col = bids_col.collection 

132 bid = BidDBModel(**await bids_col.find_single("_id", bid_id)) 

133 current_top_bid = BidDBModel( 

134 **await bids_motor_col.find_one( 

135 {BidDBModel.order_id: bid.order_id}, sort=[(BidDBModel.value, 1)] 

136 ) 

137 ) 

138 await bids_col.delete_by_id(bid_id) 

139 

140 previous_bid = await bids_motor_col.find_one( 

141 {BidDBModel.order_id: bid.order_id}, sort=[(BidDBModel.value, 1)] 

142 ) 

143 if not previous_bid: 

144 await orders_col.collection.update_one( 

145 {"id": bid.order_id}, {"$set": {OrderDBModel.best_bid: None}} 

146 ) 

147 return 

148 if current_top_bid.id != bid.id: 

149 return 

150 prev_bid_model = BidDBModel(**previous_bid) 

151 await orders_col.collection.update_one( 

152 {"id": prev_bid_model.order_id}, 

153 {"$set": {OrderDBModel.best_bid: prev_bid_model.model_dump()}}, 

154 ) 

155 order = OrderDBModel( 

156 **await orders_data_grabber.collection.find_single( 

157 "id", prev_bid_model.order_id 

158 ) 

159 ) 

160 route: dict = {} 

161 if order.stops: 

162 route["from"] = order.stops[0].address 

163 route["to"] = order.stops[-1].address 

164 

165 

166async def on_get_bids( 

167 user: SotransOIDCUserModel, params: BaseGetListQueryParams 

168) -> GenericGetListResponse[BidDBModel]: 

169 if has_role(user, SotransRole.company_director): 

170 return await bids_data_grabber.get_list(params) 

171 where = MongoDataGrabber.parse_where(params.where) 

172 if where and "order_id" in where: 

173 MongoDataGrabber.where_conversion(where, BidDBModel) 

174 assignment = await MongoDataGrabber.parse_assignment(None, user) 

175 order = await orders_col.collection.find_one( 

176 {"id": where["order_id"]} | assignment 

177 ) 

178 if order is None: 

179 raise NotFoundHTTPError("назначенный заказ") 

180 return await bids_data_grabber.get_list(params) 

181 raise BadParameterHTTPError("вы должны указать заказ")