Coverage for api/endpoints/orders/orders.py: 72%
65 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1from typing import Annotated
3from errors import log_error
4from exceptions import DocGenerationFailed
5from fastapi import APIRouter, Depends, HTTPException, Query
6from handlers.grabbers.orders import orders_data_grabber
7from handlers.orders.cancellation import on_move_to_cancel
8from handlers.orders.confirm_orders import (
9 on_apply_for_draft,
10 on_draw_up_order,
11 on_put_to_active,
12)
13from handlers.orders.exchange_orders import (
14 on_get_exchange_order_by_id,
15 on_order_through_auction,
16)
17from handlers.orders.orders import (
18 on_assign_carrier,
19 on_company_assignment,
20 on_multi_trash,
21 on_orders_update,
22 on_to_appointment,
23 on_to_exchange,
24)
25from keycloak import idp
26from sotrans_models.models._mongo import PydanticObjectId, PydanticObjectIdPath
27from sotrans_models.models.orders.order import (
28 ConfirmedOrderUpdateModel,
29 OrderDBModel,
30 OrderUpdateModel,
31 ResourceCheckOrderDBModel,
32)
33from sotrans_models.models.responses import (
34 ErrorRepr,
35 GenericGetListResponse,
36 MultiselectError,
37)
38from sotrans_models.models.roles import SotransRole
39from sotrans_models.models.users import SotransOIDCUserModel
40from starlette import status
41from utils.access_wrapper import get_active_user
42from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams
43from utils.helper import get_ids_list, multiexec
45orders_router = APIRouter(prefix="/orders", tags=["orders"])
47PydanticObjectIdListQuery = Annotated[list[PydanticObjectId], Query()]
50@orders_router.patch("/{order_id}")
51async def update_order(
52 order: OrderUpdateModel,
53 order_id: PydanticObjectIdPath,
54 _: Annotated[
55 SotransOIDCUserModel,
56 Depends(
57 idp.get_current_user(
58 required_role_names=[
59 SotransRole.company_logistician,
60 ]
61 )
62 ),
63 ],
64) -> OrderDBModel:
65 return await on_orders_update(order_id, order)
68@orders_router.put("/{order_ids}/to-exchange")
69async def orders_to_exchange(
70 order_ids: str,
71 user: Annotated[
72 SotransOIDCUserModel,
73 Depends(
74 idp.get_current_user(
75 required_role_names=[
76 SotransRole.company_logistician,
77 ]
78 )
79 ),
80 ],
81 order: OrderUpdateModel | None,
82) -> MultiselectError:
83 return await multiexec(order_ids, on_to_exchange, order, user)
86@orders_router.api_route("/{order_ids}/to-appointment")
87async def orders_to_appointment(
88 order_ids: str,
89 user: Annotated[
90 SotransOIDCUserModel,
91 Depends(
92 idp.get_current_user(
93 required_role_names=[
94 SotransRole.company_logistician,
95 ]
96 )
97 ),
98 ],
99 order: OrderUpdateModel | None,
100) -> MultiselectError:
101 return await multiexec(order_ids, on_to_appointment, order, user)
104@orders_router.put("/{order_ids}/to-canceled")
105async def move_to_canceled_order(
106 order_ids: str,
107 user: Annotated[
108 SotransOIDCUserModel,
109 Depends(
110 idp.get_current_user(
111 required_role_names=[SotransRole.company_logistician]
112 )
113 ),
114 ],
115 order: OrderUpdateModel,
116) -> MultiselectError:
117 """Accepts cancel_reason(required) and assigned(optional)"""
118 return await multiexec(order_ids, on_move_to_cancel, user, order)
121@orders_router.put(
122 "/{order_ids}/to-confirmed", responses={404: {"model": ErrorRepr}}
123)
124async def assign_carrier(
125 order_ids: str,
126 user: Annotated[
127 SotransOIDCUserModel,
128 Depends(
129 idp.get_current_user(
130 required_role_names=[SotransRole.company_logistician]
131 )
132 ),
133 ],
134 order: OrderUpdateModel,
135) -> MultiselectError:
136 """Accepts carrier(required), assigned(optional)."""
137 return await multiexec(order_ids, on_assign_carrier, user, order)
140@orders_router.put("/{order_ids}/assigned/company")
141async def assign_company(
142 order_ids: str,
143 _: Annotated[
144 SotransOIDCUserModel,
145 Depends(
146 idp.get_current_user(
147 required_role_names=[SotransRole.company_manager]
148 )
149 ),
150 ],
151 order: OrderUpdateModel,
152) -> MultiselectError:
153 return await multiexec(order_ids, on_company_assignment, order)
156@orders_router.get("")
157async def get_orders(
158 user: Annotated[
159 SotransOIDCUserModel,
160 Depends(
161 idp.get_current_user(
162 required_role_names=[SotransRole.company_logistician]
163 )
164 ),
165 ],
166 params: BaseGetListQueryParams = Depends(),
167) -> GenericGetListResponse[OrderDBModel]:
168 return await orders_data_grabber.get_list(params, user)
171@orders_router.put("/{order_ids}/to-trash")
172async def multi_trash(
173 _: Annotated[
174 SotransOIDCUserModel,
175 Depends(
176 idp.get_current_user(
177 required_role_names=[SotransRole.company_manager]
178 )
179 ),
180 ],
181 order_ids: str,
182) -> bool:
183 ids = get_ids_list(order_ids)
184 return await on_multi_trash(ids)
187@orders_router.get("/{order_id}", response_model_exclude_unset=True)
188async def get_order_by_id(
189 order_id: PydanticObjectIdPath,
190 user: Annotated[
191 SotransOIDCUserModel,
192 Depends(idp.get_current_user()),
193 ],
194 params: BaseGetOneQueryParams = Depends(),
195) -> ResourceCheckOrderDBModel:
196 return await on_get_exchange_order_by_id(
197 user,
198 order_id,
199 params,
200 )
203@orders_router.put("/{order_id}/generate-document")
204async def apply_for_draft(
205 order: ConfirmedOrderUpdateModel,
206 order_id: PydanticObjectIdPath,
207 user=Depends(get_active_user([SotransRole.carrier_logistician])),
208) -> OrderDBModel:
209 try:
210 return await on_apply_for_draft(order_id, order, user)
211 except DocGenerationFailed:
212 log_error(f"Document generation on order {order_id} failed.")
213 raise HTTPException(
214 status.HTTP_503_SERVICE_UNAVAILABLE,
215 "Генерация документов не завершена.",
216 )
219@orders_router.put("/{order_ids}/to-active")
220async def put_order_to_active(
221 order_ids: str,
222 user: Annotated[
223 SotransOIDCUserModel,
224 Depends(
225 idp.get_current_user(
226 required_role_names=[SotransRole.company_logistician]
227 )
228 ),
229 ],
230 order: OrderUpdateModel | None = None,
231) -> MultiselectError:
232 """
233 Throws Not Found on wrong subsidiary/employee, another than unverified status.
234 Accepts end_price and assigned(both are optional), but should result in valid end_price.
235 """
236 return await multiexec(order_ids, on_put_to_active, user, order)
239@orders_router.put(
240 "/{order_id}/to-unverified",
241 response_model_exclude_unset=True,
242 responses={404: {"model": ErrorRepr}},
243)
244async def draw_up_order(
245 order: ConfirmedOrderUpdateModel,
246 order_id: PydanticObjectIdPath,
247 user=Depends(get_active_user([SotransRole.carrier_logistician])),
248) -> OrderDBModel:
249 """Restricted carrier by organization_id, by confirmation_end_time, by id"""
250 return await on_draw_up_order(order_id, user, order)
253@orders_router.put(
254 "/{order_ids}/to-reserved",
255 responses={404: {"model": ErrorRepr}, 406: {"model": ErrorRepr}},
256)
257async def order_through_auction(
258 order_ids: str,
259 user: Annotated[
260 SotransOIDCUserModel,
261 Depends(
262 idp.get_current_user(
263 required_role_names=[
264 SotransRole.company_logistician,
265 SotransRole.bid_service,
266 ]
267 )
268 ),
269 ],
270 order: OrderUpdateModel | None = None,
271) -> MultiselectError:
272 return await multiexec(order_ids, on_order_through_auction, user, order)