Coverage for api/endpoints/orders/appointment.py: 81%

1from typing import Annotated 

2 

3from fastapi import APIRouter 

4from fastapi.params import Depends 

5from handlers.authorization.company_employee_access import ( 

6 assigned_subsidiary_or_logistician_for_order_query, 

7) 

8from handlers.grabbers.orders import orders_data_grabber 

9from handlers.orders.to_appointment import ( 

10 on_appointment_to_exchange, 

11 on_get_whole_appointment, 

12 on_put_appointment_to_buffer, 

13 on_put_to_trash, 

14) 

15from keycloak import idp 

16from sotrans_models.models._mongo import PydanticObjectIdPath 

17from sotrans_models.models.orders.order import OrderDBModel, OrderUpdateModel 

18from sotrans_models.models.responses import ( 

19 ErrorRepr, 

20 GenericGetListResponse, 

21 MultiselectError, 

22) 

23from sotrans_models.models.roles import SotransRole 

24from sotrans_models.models.users import SotransOIDCUserModel 

25from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

26from utils.helper import multiexec 

27 

28appointment_router = APIRouter( 

29 prefix="/appointment-orders", tags=["orders", "appointment"] 

30) 

31 

32 

33@appointment_router.put( 

34 "/{order_ids}/to-trash", 

35 responses={404: {"model": ErrorRepr}}, 

36) 

37async def move_appointed_to_trash( 

38 order_ids: str, 

39 user: Annotated[ 

40 SotransOIDCUserModel, 

41 Depends( 

42 idp.get_current_user( 

43 required_role_names=[SotransRole.company_logistician] 

44 ) 

45 ), 

46 ], 

47) -> MultiselectError: 

48 return await multiexec(order_ids, on_put_to_trash, user) 

49 

50 

51@appointment_router.put( 

52 "/{order_ids}/to-buffer", responses={404: {"model": ErrorRepr}} 

53) 

54async def appointment_to_buffer( 

55 order_ids: str, 

56 user: Annotated[ 

57 SotransOIDCUserModel, 

58 Depends( 

59 idp.get_current_user( 

60 required_role_names=[SotransRole.company_logistician] 

61 ) 

62 ), 

63 ], 

64 order: OrderUpdateModel | None = None, 

65) -> MultiselectError: 

66 return await multiexec( 

67 order_ids, on_put_appointment_to_buffer, user, order 

68 ) 

69 

70 

71@appointment_router.put("/{order_ids}/to-exchange") 

72async def appointment_to_exchange( 

73 order_ids: str, 

74 user: Annotated[ 

75 SotransOIDCUserModel, 

76 Depends( 

77 idp.get_current_user( 

78 required_role_names=[SotransRole.company_logistician] 

79 ) 

80 ), 

81 ], 

82 order: OrderUpdateModel, 

83) -> MultiselectError: 

84 """Accepts start_price(required) and assigned. Must be finished with auction_end_time""" 

85 return await multiexec(order_ids, on_appointment_to_exchange, user, order) 

86 

87 

88@appointment_router.get("") 

89async def get_appointment( 

90 user: Annotated[ 

91 SotransOIDCUserModel, 

92 Depends( 

93 idp.get_current_user( 

94 required_role_names=[SotransRole.company_logistician] 

95 ) 

96 ), 

97 ], 

98 params: BaseGetListQueryParams = Depends(), 

99) -> GenericGetListResponse[OrderDBModel]: 

100 return await on_get_whole_appointment(user, params) 

101 

102 

103@appointment_router.get("/{order_id}", responses={404: {"model": ErrorRepr}}) 

104async def get_appointment_by_id( 

105 order_id: PydanticObjectIdPath, 

106 user: Annotated[ 

107 SotransOIDCUserModel, 

108 Depends( 

109 idp.get_current_user( 

110 required_role_names=[SotransRole.company_logistician] 

111 ) 

112 ), 

113 ], 

114 params: BaseGetOneQueryParams = Depends(), 

115) -> OrderDBModel: 

116 restriction_q = assigned_subsidiary_or_logistician_for_order_query(user) 

117 return await orders_data_grabber.get_one_by_id_with_pattern( 

118 order_id, params, restriction_q 

119 )