Coverage for api/endpoints/misc/clients.py: 49%

1from typing import Annotated 

2 

3import motor 

4from database.updater import remove_client_from_deps 

5from exceptions import NoAccessHTTPError, NotFoundHTTPError 

6from fastapi import APIRouter, Depends 

7from handlers.authorization.check_role import has_role 

8from handlers.authorization.company_employee_access import ( 

9 get_assigned_clients_ids, 

10) 

11from handlers.grabbers.clients import clients_grabber 

12from handlers.misc.clients import ( 

13 assign_manager_to_client, 

14 on_assign_subsidiary_to_client, 

15 on_client_revoke_manager, 

16 on_create_client, 

17 on_get_clients_managers, 

18 on_patch_client, 

19 on_revoke_subsidiary, 

20) 

21from keycloak import idp 

22from mongodb import clients_col, cm_clients_col, subsidiaries_clients_col 

23from sotrans_models.models._base import InsertByOIDModel, InsertByUUIDModel 

24from sotrans_models.models._mongo import PydanticObjectIdPath 

25from sotrans_models.models.misc.client import ( 

26 AssignedToClientResponse, 

27 ClientCreateModel, 

28 ClientDBModel, 

29 ClientUpdateModel, 

30) 

31from sotrans_models.models.responses import ErrorRepr, GenericGetListResponse 

32from sotrans_models.models.roles import SotransRole 

33from sotrans_models.models.users import ( 

34 SotransOIDCUserModel, 

35 SotransUserDBModel, 

36) 

37from starlette import status 

38from utils.data_grabber import ( 

39 BaseGetListQueryParams, 

40 BaseGetOneQueryParams, 

41 adjust_search_query, 

42) 

43from utils.helper import etag_detalizer 

44 

45clients_router = APIRouter( 

46 prefix="/clients", 

47 tags=["clients"], 

48) 

49 

50clients_assignment_router = APIRouter( 

51 prefix="/clients", 

52 tags=["managers_clients_assignment"], 

53) 

54 

55 

56@clients_router.post("", status_code=status.HTTP_201_CREATED) 

57async def create_client( 

58 client: ClientCreateModel, 

59 _: Annotated[ 

60 SotransOIDCUserModel, 

61 Depends( 

62 idp.get_current_user( 

63 required_role_names=[SotransRole.company_director] 

64 ) 

65 ), 

66 ], 

67) -> ClientDBModel: 

68 return await on_create_client(client) 

69 

70 

71@clients_router.patch("/{client_id}") 

72async def change_client_data( 

73 client_id: PydanticObjectIdPath, 

74 client: ClientUpdateModel, 

75 _: Annotated[ 

76 SotransOIDCUserModel, 

77 Depends( 

78 idp.get_current_user( 

79 required_role_names=[SotransRole.company_director] 

80 ) 

81 ), 

82 ], 

83) -> ClientDBModel: 

84 return await on_patch_client(client_id, client) 

85 

86 

87@clients_router.delete( 

88 "/{client_id}", 

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

90 status_code=204, 

91) 

92async def remove_client( 

93 client_id: PydanticObjectIdPath, 

94 _: Annotated[ 

95 SotransOIDCUserModel, 

96 Depends( 

97 idp.get_current_user( 

98 required_role_names=[SotransRole.company_director] 

99 ) 

100 ), 

101 ], 

102 resource_with_etag: ClientUpdateModel | None = None, 

103): 

104 clients_motor_col: motor.MotorCollection = ( 

105 clients_grabber.collection.collection 

106 ) 

107 etag_q = ( 

108 {"etag": resource_with_etag.etag} 

109 if resource_with_etag is not None 

110 else {} 

111 ) 

112 deletion = await clients_motor_col.delete_one({"_id": client_id} | etag_q) 

113 if deletion.deleted_count == 0: 

114 await etag_detalizer(clients_col, etag_q, {"_id": client_id}) 

115 raise NotFoundHTTPError("клиент") 

116 await subsidiaries_clients_col.collection.update_many( 

117 {}, {"$pull": {"clients_ids": client_id}} 

118 ) 

119 await cm_clients_col.collection.update_many( 

120 {}, {"$pull": {"clients_ids": client_id}} 

121 ) 

122 await remove_client_from_deps(client_id) 

123 

124 

125@clients_router.get("/{client_id}", responses={404: {"model": ErrorRepr}}) 

126async def clients_details( 

127 client_id: PydanticObjectIdPath, 

128 user: Annotated[ 

129 SotransOIDCUserModel, 

130 Depends( 

131 idp.get_current_user( 

132 required_role_names=[SotransRole.company_manager] 

133 ) 

134 ), 

135 ], 

136 params: BaseGetOneQueryParams = Depends(), 

137) -> ClientDBModel: 

138 if not has_role(user, SotransRole.company_director): 

139 assigned = await get_assigned_clients_ids(user) 

140 if client_id not in assigned: 

141 raise NoAccessHTTPError("клиент не назначен") 

142 return await clients_grabber.get_one(client_id, params=params) 

143 

144 

145@clients_router.get("") 

146async def get_clients_list( 

147 user: Annotated[ 

148 SotransOIDCUserModel, 

149 Depends( 

150 idp.get_current_user( 

151 required_role_names=[SotransRole.company_manager] 

152 ) 

153 ), 

154 ], 

155 params: BaseGetListQueryParams = Depends(), 

156) -> GenericGetListResponse[ClientDBModel]: 

157 if not has_role(user, SotransRole.company_director): 

158 assigned_clients = await get_assigned_clients_ids(user) 

159 params.where = adjust_search_query( 

160 params.where, 

161 "id", 

162 {str(ci) for ci in assigned_clients}, 

163 ) 

164 return await clients_grabber.get_list(params=params) 

165 

166 

167# /-----------------------------------------------------------------------/ 

168# below assignment routes are placed 

169# /-----------------------------------------------------------------------/ 

170 

171 

172@clients_assignment_router.put( 

173 "/{client_id}/assign", status_code=status.HTTP_201_CREATED 

174) 

175async def assign_to_client( 

176 client_id: PydanticObjectIdPath, 

177 assignee: InsertByOIDModel | InsertByUUIDModel, 

178 _: Annotated[ 

179 SotransOIDCUserModel, 

180 Depends( 

181 idp.get_current_user( 

182 required_role_names=[SotransRole.company_director] 

183 ) 

184 ), 

185 ], 

186) -> GenericGetListResponse[SotransUserDBModel] | AssignedToClientResponse: 

187 if isinstance(assignee, InsertByOIDModel): 

188 return await on_assign_subsidiary_to_client(assignee.id, client_id) 

189 return await assign_manager_to_client(client_id, assignee.id) 

190 

191 

192@clients_assignment_router.delete( 

193 "/{client_id}/revoke", status_code=status.HTTP_204_NO_CONTENT 

194) 

195async def revoke_from_client( 

196 client_id: PydanticObjectIdPath, 

197 assignee: InsertByOIDModel | InsertByUUIDModel, 

198 _: Annotated[ 

199 SotransOIDCUserModel, 

200 Depends( 

201 idp.get_current_user( 

202 required_role_names=[SotransRole.company_director] 

203 ) 

204 ), 

205 ], 

206): 

207 if isinstance(assignee, InsertByOIDModel): 

208 return await on_revoke_subsidiary(assignee.id, client_id) 

209 await on_client_revoke_manager(client_id, assignee.id) 

210 

211 

212@clients_assignment_router.get("/{client_id}/assign") 

213async def get_clients_assigned( 

214 client_id: PydanticObjectIdPath, 

215 _: Annotated[ 

216 SotransOIDCUserModel, 

217 Depends( 

218 idp.get_current_user( 

219 required_role_names=[SotransRole.company_logistician] 

220 ) 

221 ), 

222 ], 

223) -> GenericGetListResponse[SotransUserDBModel]: 

224 return await on_get_clients_managers(client_id)