Coverage for api/endpoints/resources/trucks.py: 81%

1from fastapi import APIRouter, Depends 

2from handlers.resources.trucks import ( 

3 on_create_truck, 

4 on_delete_truck, 

5 on_get_truck, 

6 on_get_trucks, 

7 on_update_truck, 

8) 

9from keycloak import idp 

10from sotrans_models.body_props import BODY_TYPES_MATCH, LOADING_TYPES_MATCH 

11from sotrans_models.models._mongo import PydanticObjectIdPath 

12from sotrans_models.models.resources.trucks import ( 

13 TruckCreateModel, 

14 TruckDBModel, 

15 TruckUpdateModel, 

16) 

17from sotrans_models.models.responses import GenericGetListResponse 

18from sotrans_models.models.roles import SotransRole 

19from sotrans_models.models.users import SotransOIDCUserModel 

20from starlette import status 

21from typing_extensions import Annotated 

22from utils.data_grabber import BaseGetListQueryParams 

23 

24trucks_router = APIRouter(prefix="/trucks", tags=["trucks"]) 

25 

26 

27@trucks_router.post("", status_code=status.HTTP_201_CREATED) 

28async def create_truck( 

29 creation_data: TruckCreateModel, 

30 executor: Annotated[ 

31 SotransOIDCUserModel, 

32 Depends( 

33 idp.get_current_user( 

34 required_role_names=[SotransRole.carrier_logistician] 

35 ) 

36 ), 

37 ], 

38) -> TruckDBModel: 

39 return await on_create_truck(creation_data, executor) 

40 

41 

42@trucks_router.get("") 

43async def get_trucks( 

44 _: Annotated[ 

45 SotransOIDCUserModel, 

46 Depends( 

47 idp.get_current_user( 

48 required_role_names=[SotransRole.company_logistician] 

49 ) 

50 ), 

51 ], 

52 params: BaseGetListQueryParams = Depends(), 

53) -> GenericGetListResponse[TruckDBModel]: 

54 return await on_get_trucks(params) 

55 

56 

57@trucks_router.get("/properties") 

58async def get_types_mapping( 

59 _: Annotated[ 

60 SotransOIDCUserModel, 

61 Depends(idp.get_current_user([SotransRole.company_logistician])), 

62 ], 

63) -> dict[str, list[dict[str, str]]]: 

64 return { 

65 "body_types": [ 

66 {"value": BODY_TYPES_MATCH[k].value, "label": k} 

67 for k in BODY_TYPES_MATCH 

68 ], 

69 "loading_types": [ 

70 {"value": LOADING_TYPES_MATCH[k].value, "label": k} 

71 for k in LOADING_TYPES_MATCH 

72 ], 

73 } 

74 

75 

76@trucks_router.get("/{truck_id}") 

77async def get_truck( 

78 truck_id: PydanticObjectIdPath, 

79 executor: Annotated[SotransOIDCUserModel, Depends(idp.get_current_user())], 

80) -> TruckDBModel: 

81 return await on_get_truck(truck_id, executor) 

82 

83 

84@trucks_router.patch("/{truck_id}") 

85async def update_truck( 

86 truck_id: PydanticObjectIdPath, 

87 update_data: TruckUpdateModel, 

88 executor: Annotated[ 

89 SotransOIDCUserModel, 

90 Depends( 

91 idp.get_current_user( 

92 required_role_names=[SotransRole.carrier_logistician] 

93 ) 

94 ), 

95 ], 

96) -> TruckDBModel: 

97 return await on_update_truck(truck_id, update_data, executor) 

98 

99 

100@trucks_router.delete( 

101 "/{truck_id}", 

102 responses={204: {"description": "No content"}}, 

103 status_code=204, 

104) 

105async def delete_truck( 

106 truck_id: PydanticObjectIdPath, 

107 executor: Annotated[ 

108 SotransOIDCUserModel, 

109 Depends(idp.get_current_user([SotransRole.carrier_logistician])), 

110 ], 

111 resource_with_etag: TruckUpdateModel | None = None, 

112): 

113 await on_delete_truck( 

114 truck_id, 

115 executor, 

116 resource_with_etag.etag if resource_with_etag is not None else None, 

117 )