Coverage for api/endpoints/orders/active_orders.py: 59%

1from fastapi import APIRouter, Depends 

2from handlers.authorization.check_role import has_role 

3from handlers.authorization.company_employee_access import ( 

4 flexible_company_assertion_query, 

5) 

6from handlers.authorization.confidential import hide_model_fields_from_carrier 

7from handlers.grabbers.orders import orders_data_grabber 

8from handlers.orders.active_orders import ( 

9 on_get_actives, 

10 on_get_stages_suggestions, 

11 on_update_current_stage, 

12) 

13from sotrans_models.models._mongo import PydanticObjectIdPath 

14from sotrans_models.models.orders.order import ( 

15 OrderDBModel, 

16 OrderStageSubmitModel, 

17 OrderStageSuggestModel, 

18 OrderStatus, 

19) 

20from sotrans_models.models.responses import ErrorRepr, GenericGetListResponse 

21from sotrans_models.models.roles import SotransRole 

22from utils.access_wrapper import get_active_user 

23from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

24from utils.helper import get_org_oid 

25 

26active_orders_router = APIRouter( 

27 prefix="/active-orders", tags=["orders", "active"] 

28) 

29 

30 

31@active_orders_router.get("/{order_id}/stages") 

32async def get_stages_suggestions( 

33 order_id: PydanticObjectIdPath, 

34 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

35) -> list[OrderStageSuggestModel]: 

36 return await on_get_stages_suggestions(order_id, user) 

37 

38 

39@active_orders_router.get("", response_model_exclude_unset=True) 

40async def get_active_orders( 

41 user=Depends( 

42 get_active_user( 

43 [ 

44 SotransRole.carrier_logistician, 

45 SotransRole.drivers_bot, 

46 ] 

47 ) 

48 ), 

49 params: BaseGetListQueryParams = Depends(), 

50) -> GenericGetListResponse[OrderDBModel]: 

51 return await on_get_actives(user, params) 

52 

53 

54@active_orders_router.put("/{order_id}/current_stage") 

55async def update_current_stage( 

56 order_id: PydanticObjectIdPath, 

57 stage: OrderStageSubmitModel, 

58 user=Depends( 

59 get_active_user( 

60 [ 

61 SotransRole.carrier_logistician, 

62 SotransRole.drivers_bot, 

63 ] 

64 ) 

65 ), 

66) -> OrderDBModel: 

67 return await on_update_current_stage(order_id, stage, user) 

68 

69 

70@active_orders_router.get( 

71 "/{order_id}", 

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

73 response_model_exclude_unset=True, 

74) 

75async def get_active_by_id( 

76 order_id: PydanticObjectIdPath, 

77 user=Depends( 

78 get_active_user( 

79 [SotransRole.carrier_logistician, SotransRole.drivers_bot] 

80 ) 

81 ), 

82 params: BaseGetOneQueryParams = Depends(), 

83) -> OrderDBModel: 

84 status_restriction = {OrderDBModel.status: OrderStatus.active.value} 

85 if has_role(user, SotransRole.company_director) or has_role( 

86 user, SotransRole.drivers_bot 

87 ): 

88 return await orders_data_grabber.get_one_by_id_with_pattern( 

89 order_id, params, status_restriction 

90 ) 

91 if has_role(user, SotransRole.company_logistician): 

92 restriction_q = await flexible_company_assertion_query(user) 

93 return await orders_data_grabber.get_one_by_id_with_pattern( 

94 order_id, params, restriction_q | status_restriction 

95 ) 

96 

97 org_id = get_org_oid(user) 

98 

99 return await orders_data_grabber.get_one_by_id_with_pattern( 

100 order_id, 

101 params, 

102 status_restriction | {OrderDBModel.carrier.id: org_id}, 

103 [hide_model_fields_from_carrier], 

104 )