Coverage for api/endpoints/orders/exchange_orders.py: 75%
57 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1import aiofiles.os as aios
2from fastapi import APIRouter, Depends
3from handlers.orders.bids import on_make_bid
4from handlers.orders.exchange_orders import (
5 export_order,
6 get_order_with_resource_check,
7 on_confirm_order,
8 on_exchange_update,
9 on_get_exchange,
10 on_get_exchange_order_by_id,
11 on_get_my_bids,
12 on_order_through_auction,
13 on_order_to_appointment,
14 on_put_exchange_to_buffer,
15 on_put_exchange_to_trash,
16)
17from keycloak import idp
18from sotrans_models.models._mongo import PydanticObjectIdPath
19from sotrans_models.models.orders.bid import BidCreateModel, BidDBModel
20from sotrans_models.models.orders.order import (
21 ExchangeOrderUpdateModel,
22 OrderDBModel,
23 OrderUpdateModel,
24 ResourceCheckOrderDBModel,
25)
26from sotrans_models.models.responses import (
27 ErrorRepr,
28 GenericGetListResponse,
29 MultiselectError,
30)
31from sotrans_models.models.roles import SotransRole
32from sotrans_models.models.users import SotransOIDCUserModel
33from starlette import status
34from starlette.background import BackgroundTasks
35from starlette.responses import FileResponse
36from typing_extensions import Annotated
37from utils.access_wrapper import get_active_user
38from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams
39from utils.helper import multiexec
41exchange_orders_router = APIRouter(
42 prefix="/exchange-orders", tags=["exchange", "orders"]
43)
46@exchange_orders_router.get("/xlsx")
47async def excel_export_exchange(
48 user: Annotated[
49 SotransOIDCUserModel,
50 Depends(
51 get_active_user(
52 [SotransRole.carrier_logistician],
53 company_security_verification=None,
54 )
55 ),
56 ],
57 bg_task: BackgroundTasks,
58 params: BaseGetListQueryParams = Depends(),
59) -> FileResponse:
60 path = await export_order(user, params)
61 bg_task.add_task(aios.remove, path)
62 return FileResponse(path)
65@exchange_orders_router.put(
66 "/{order_ids}/to-buffer", responses={404: {"model": ErrorRepr}}
67)
68async def put_order_to_buffer(
69 order_ids: str,
70 _: Annotated[
71 SotransOIDCUserModel,
72 Depends(
73 idp.get_current_user(
74 required_role_names=[SotransRole.company_logistician]
75 )
76 ),
77 ],
78 order: OrderUpdateModel | None = None,
79) -> MultiselectError:
80 return await multiexec(order_ids, on_put_exchange_to_buffer, order)
83@exchange_orders_router.patch(
84 "/{order_id}", responses={404: {"model": ErrorRepr}}
85)
86async def update_order(
87 order: ExchangeOrderUpdateModel,
88 user: Annotated[
89 SotransOIDCUserModel,
90 Depends(
91 idp.get_current_user(
92 required_role_names=[SotransRole.company_logistician]
93 )
94 ),
95 ],
96 order_id: PydanticObjectIdPath,
97) -> OrderDBModel:
98 # todo remove this endpoint as duplicate(with update in 'orders'?)
99 return await on_exchange_update(order_id, order, user)
102@exchange_orders_router.put(
103 "/{order_ids}/to-trash",
104 responses={204: {"description": "No content"}, 404: {"model": ErrorRepr}},
105)
106async def remove_order_from_exchange(
107 order_ids: str,
108 user: Annotated[
109 SotransOIDCUserModel,
110 Depends(
111 idp.get_current_user(
112 required_role_names=[SotransRole.company_manager]
113 )
114 ),
115 ],
116) -> MultiselectError:
117 return await multiexec(order_ids, on_put_exchange_to_trash, user)
120@exchange_orders_router.get("")
121async def get_exchange(
122 user=Depends(
123 get_active_user(
124 [SotransRole.carrier_logistician],
125 company_security_verification=None,
126 )
127 ),
128 params: BaseGetListQueryParams = Depends(),
129) -> GenericGetListResponse[OrderDBModel]:
130 """For any status starting with 'exchange' for CompLogistician"""
131 return await on_get_exchange(user, params)
134@exchange_orders_router.get("/my-bids")
135async def get_exchange_bids(
136 user=Depends(
137 get_active_user(
138 [SotransRole.carrier_logistician],
139 company_security_verification=None,
140 )
141 ),
142 params: BaseGetListQueryParams = Depends(),
143) -> GenericGetListResponse[OrderDBModel]:
144 """For any status starting with 'exchange' for CompLogistician"""
145 return await on_get_my_bids(user, params)
148@exchange_orders_router.put(
149 "/{order_ids}/to-reserved",
150 responses={404: {"model": ErrorRepr}, 406: {"model": ErrorRepr}},
151)
152async def order_through_auction(
153 order_ids: str,
154 user: Annotated[
155 SotransOIDCUserModel,
156 Depends(
157 idp.get_current_user(
158 required_role_names=[
159 SotransRole.company_logistician,
160 SotransRole.bid_service,
161 ]
162 )
163 ),
164 ],
165 order: OrderUpdateModel | None = None,
166) -> MultiselectError:
167 return await multiexec(order_ids, on_order_through_auction, user, order)
170@exchange_orders_router.put(
171 "/{order_id}/bid",
172 status_code=status.HTTP_201_CREATED,
173 responses={
174 400: {"model": ErrorRepr},
175 404: {"model": ErrorRepr},
176 406: {"model": ErrorRepr},
177 },
178 response_model_exclude={"text_search"},
179)
180async def bid_for_order(
181 order_id: PydanticObjectIdPath,
182 bid: BidCreateModel,
183 user=Depends(get_active_user([SotransRole.carrier_logistician])),
184) -> BidDBModel:
185 return await on_make_bid(user, bid, order_id)
188@exchange_orders_router.put("/{order_ids}/to-confirmed")
189async def confirm_order(
190 order_ids: str,
191 user: Annotated[
192 SotransOIDCUserModel,
193 Depends(
194 idp.get_current_user(
195 required_role_names=[SotransRole.company_logistician]
196 )
197 ),
198 ],
199 order: OrderUpdateModel | None = None,
200) -> MultiselectError:
201 """Accepts carrier(optional) and assigned(optional)."""
202 return await multiexec(order_ids, on_confirm_order, user, order)
205@exchange_orders_router.get(
206 "/{order_id}/is-admitted",
207 responses={404: {"model": ErrorRepr}},
208 response_model_exclude_unset=True,
209)
210async def get_exchange_order_by_id_with_verification(
211 order_id: PydanticObjectIdPath,
212 user=Depends(get_active_user([SotransRole.carrier_logistician])),
213 params: BaseGetOneQueryParams = Depends(),
214) -> ResourceCheckOrderDBModel:
215 return await get_order_with_resource_check(order_id, params, user)
218@exchange_orders_router.get(
219 "/{order_id}",
220 responses={404: {"model": ErrorRepr}},
221 response_model_exclude_unset=True,
222)
223async def get_exchange_order_by_id(
224 order_id: PydanticObjectIdPath,
225 user=Depends(
226 get_active_user(
227 [SotransRole.carrier_logistician, SotransRole.bid_service]
228 )
229 ),
230 params: BaseGetOneQueryParams = Depends(),
231) -> OrderDBModel:
232 """For any status starting with 'exchange' in exchange collection"""
233 return await on_get_exchange_order_by_id(user, order_id, params)
236@exchange_orders_router.put("/{order_ids}/to-appointment")
237async def order_to_appointment(
238 order_ids: str,
239 user: Annotated[
240 SotransOIDCUserModel,
241 Depends(
242 idp.get_current_user(
243 required_role_names=[SotransRole.company_logistician]
244 )
245 ),
246 ],
247 order: OrderUpdateModel | None = None,
248) -> MultiselectError:
249 return await multiexec(order_ids, on_order_to_appointment, user, order)