Coverage for api/endpoints/users.py: 77%

1from typing import Annotated 

2 

3from fastapi import APIRouter, Depends 

4from handlers import users 

5from handlers.users import ( 

6 keycloak_webhook_handler, 

7 on_get_assigned_orders_to_employee, 

8 on_get_assigned_to_employee_clients, 

9 update_users_roles, 

10) 

11from keycloak import idp 

12from sotrans_fastapi_keycloak._uuid import PydanticUUIDPath 

13from sotrans_models.models.misc.client import ClientDBModel 

14from sotrans_models.models.orders.order import OrderDBModel 

15from sotrans_models.models.responses import ErrorRepr, GenericGetListResponse 

16from sotrans_models.models.roles import SotransRole 

17from sotrans_models.models.users import ( 

18 CreateUserWithRoleModel, 

19 SotransKeycloakUserModel, 

20 SotransKeycloakUserPublicUpdateModel, 

21 SotransKeycloakUserUpdateModel, 

22 SotransOIDCUserModel, 

23 SotransUserDBModel, 

24 UpdateRoleModel, 

25) 

26from starlette import status 

27from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

28 

29users_router = APIRouter(prefix="/users", tags=["users"]) 

30 

31 

32@users_router.post("", status_code=status.HTTP_201_CREATED) 

33async def create_user( 

34 creation_data: CreateUserWithRoleModel, 

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

36) -> SotransUserDBModel: 

37 return await users.on_create_users(creation_data, executor) 

38 

39 

40@users_router.get("", responses={400: {"model": ErrorRepr}}) 

41async def get_users( 

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

43 params: BaseGetListQueryParams = Depends(), 

44) -> GenericGetListResponse[SotransUserDBModel]: 

45 return await users.on_get_users(executor, params) 

46 

47 

48@users_router.get("/{user_id}", responses={404: {"model": ErrorRepr}}) 

49async def get_user_by_id( 

50 user_id: PydanticUUIDPath, 

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

52 params: BaseGetOneQueryParams = Depends(), 

53) -> SotransUserDBModel: 

54 return await users.on_get_user(user_id, executor, params) 

55 

56 

57@users_router.delete( 

58 "/{user_id}", 

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

60 status_code=204, 

61) 

62async def delete_user( 

63 user_id: str, 

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

65): 

66 await users.on_delete_user(user_id, executor) 

67 

68 

69@users_router.patch("/{user_id}/roles") 

70async def update_user_roles( 

71 user_id: str, 

72 role_model: UpdateRoleModel, 

73 executor: Annotated[ 

74 SotransOIDCUserModel, 

75 Depends( 

76 idp.get_current_user( 

77 required_role_names=[SotransRole.carrier_director] 

78 ) 

79 ), 

80 ], 

81) -> UpdateRoleModel: 

82 return await update_users_roles(user_id, executor, role_model.role) 

83 

84 

85@users_router.patch("/{user_id}") 

86async def update_user( 

87 user_id: str, 

88 update_data: SotransKeycloakUserUpdateModel 

89 | SotransKeycloakUserPublicUpdateModel, 

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

91) -> SotransKeycloakUserModel | SotransUserDBModel: 

92 return await users.on_update_user(user_id, update_data, executor) 

93 

94 

95@users_router.patch("/{user_id}/subsidiary") 

96async def update_subsidiary( 

97 user_id: str, 

98 sub_id: str, 

99 _: Annotated[ 

100 SotransOIDCUserModel, 

101 Depends( 

102 idp.get_current_user( 

103 required_role_names=[SotransRole.carrier_director] 

104 ) 

105 ), 

106 ], 

107) -> SotransKeycloakUserModel: 

108 return await users.on_update_subsidiary(user_id, sub_id) 

109 

110 

111@users_router.post("/webhook") 

112async def keycloak_webhook(data: dict): 

113 await keycloak_webhook_handler(data) 

114 

115 

116@users_router.get("/{user_id}/clients") 

117async def get_assigned_to_employee_clients( 

118 _: Annotated[ 

119 SotransOIDCUserModel, 

120 Depends( 

121 idp.get_current_user( 

122 required_role_names=[SotransRole.company_manager] 

123 ) 

124 ), 

125 ], 

126 user_id: PydanticUUIDPath, 

127 params: BaseGetListQueryParams = Depends(), 

128) -> GenericGetListResponse[ClientDBModel]: 

129 return await on_get_assigned_to_employee_clients(user_id, params) 

130 

131 

132@users_router.get("/{user_id}/orders") 

133async def get_assigned_orders_to_employees( 

134 user_id: PydanticUUIDPath, 

135 _: Annotated[ 

136 SotransOIDCUserModel, 

137 Depends( 

138 idp.get_current_user( 

139 required_role_names=[SotransRole.company_manager] 

140 ) 

141 ), 

142 ], 

143 params: BaseGetListQueryParams = Depends(), 

144) -> GenericGetListResponse[OrderDBModel]: 

145 return await on_get_assigned_orders_to_employee(user_id, params)