confirm_order_and_return(
order_id: ObjectId,
carrier_data: dict[str, Any],
best_bid: dict | None = None,
restriction_query: dict[str, Any] | None = None,
update_data: dict | None = None,
)
@:param carrier_data - data directly from database
Source code in api/database/orders.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69 | async def confirm_order_and_return(
order_id: ObjectId,
carrier_data: dict[str, Any],
best_bid: dict | None = None,
restriction_query: dict[str, Any] | None = None,
update_data: dict | None = None,
):
"""@:param carrier_data - data directly from database"""
if not update_data:
update_data = {}
update = {
OrderDBModel.status: OrderStatus.confirmed.value,
OrderDBModel.confirmation_end_time: get_current_datetime()
+ timedelta(minutes=await get_confirmation_interval()),
OrderDBModel.carrier: carrier_data,
}
if best_bid is not None:
update[OrderDBModel.best_bid] = best_bid
update[OrderDBModel.end_price] = best_bid[BidDBModel.value]
if restriction_query is None:
restriction_query = {}
order = await orders_col.collection.find_one_and_update(
restriction_query | {"id": order_id},
{"$set": update_data | update},
return_document=ReturnDocument.AFTER,
)
if order is None:
etag_container = copy.deepcopy(restriction_query)
restriction_query.pop("etag", None)
await etag_detalizer(
orders_col, etag_container, restriction_query | {"id": order_id}
)
raise NotFoundHTTPError("заказ")
await update_etag_and_text_search(
order, orders_col, OrderDBModel, get_orders_text_search
)
return order
|