Coverage for api/endpoints/organization/subsidiaries.py: 64%

1from typing import Annotated 

2 

3from exceptions import NotFoundHTTPError 

4from fastapi import APIRouter, Depends 

5from handlers.grabbers.orders import orders_data_grabber 

6from handlers.grabbers.organizations import subsidiary_grabber 

7from handlers.subsidiaries import ( 

8 on_add_client_to_subsidiary, 

9 on_create_subsidiary, 

10 on_get_assigned_clients_full_by_sub, 

11 on_remove_subsidiary, 

12 on_update_subsidiary, 

13 revoke_subsidiary, 

14) 

15from keycloak import idp 

16from mongodb import subsidiaries_col 

17from sotrans_models.models._mongo import ( 

18 PydanticObjectIdPath, 

19 PydanticObjectIdQuery, 

20) 

21from sotrans_models.models.misc.client import ( 

22 ClientDBModel, 

23 SubsidiaryClientsModel, 

24) 

25from sotrans_models.models.orders.order import OrderDBModel 

26from sotrans_models.models.organizations import ( 

27 CreateSubsidiaryModel, 

28 SubsidiaryDBModel, 

29 UpdateSubsidiaryModel, 

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 update_search_query, 

42) 

43 

44subsidiary_router = APIRouter(prefix="/subsidiary", tags=["subsidiaries"]) 

45 

46 

47@subsidiary_router.post("", status_code=status.HTTP_201_CREATED) 

48async def create_subsidiary( 

49 _: Annotated[ 

50 SotransOIDCUserModel, 

51 Depends( 

52 idp.get_current_user( 

53 required_role_names=[SotransRole.company_director] 

54 ) 

55 ), 

56 ], 

57 subdivision: CreateSubsidiaryModel, 

58) -> SubsidiaryDBModel: 

59 return await on_create_subsidiary(subdivision) 

60 

61 

62@subsidiary_router.patch("/{sub_id}", responses={404: {"model": ErrorRepr}}) 

63async def update_subsidiary( 

64 sub_id: PydanticObjectIdPath, 

65 subdivision: UpdateSubsidiaryModel, 

66 _: Annotated[ 

67 SotransOIDCUserModel, 

68 Depends( 

69 idp.get_current_user( 

70 required_role_names=[SotransRole.company_director] 

71 ) 

72 ), 

73 ], 

74) -> SubsidiaryDBModel: 

75 return await on_update_subsidiary(sub_id, subdivision) 

76 

77 

78@subsidiary_router.delete("/{sub_id}", status_code=status.HTTP_204_NO_CONTENT) 

79async def remove_subsidiary( 

80 sub_id: PydanticObjectIdPath, 

81 _: Annotated[ 

82 SotransOIDCUserModel, 

83 Depends( 

84 idp.get_current_user( 

85 required_role_names=[SotransRole.company_director] 

86 ) 

87 ), 

88 ], 

89 resource_with_etag: UpdateSubsidiaryModel | None = None, 

90): 

91 await on_remove_subsidiary( 

92 sub_id, 

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

94 ) 

95 

96 

97@subsidiary_router.get("") 

98async def get_subsidiaries( 

99 _: Annotated[ 

100 SotransOIDCUserModel, 

101 Depends( 

102 idp.get_current_user( 

103 required_role_names=[SotransRole.company_logistician] 

104 ) 

105 ), 

106 ], 

107 params: BaseGetListQueryParams = Depends(), 

108) -> GenericGetListResponse[SubsidiaryDBModel]: 

109 return await subsidiary_grabber.get_list(params) 

110 

111 

112@subsidiary_router.put("/{subsidiary_id}/clients") 

113async def assign_subsidiary_to_client( 

114 subsidiary_id: PydanticObjectIdPath, 

115 client_id: PydanticObjectIdQuery, 

116 _: Annotated[ 

117 SotransOIDCUserModel, 

118 Depends( 

119 idp.get_current_user( 

120 required_role_names=[SotransRole.company_director] 

121 ) 

122 ), 

123 ], 

124) -> SubsidiaryClientsModel: 

125 return await on_add_client_to_subsidiary(subsidiary_id, client_id) 

126 

127 

128@subsidiary_router.delete("/{sub_id}/clients") 

129async def revoke_client_from_subsidiary( 

130 sub_id: PydanticObjectIdPath, 

131 client_id: PydanticObjectIdQuery, 

132 _: Annotated[ 

133 SotransOIDCUserModel, 

134 Depends( 

135 idp.get_current_user( 

136 required_role_names=[SotransRole.company_director] 

137 ) 

138 ), 

139 ], 

140) -> SubsidiaryClientsModel: 

141 return await revoke_subsidiary(sub_id, client_id) 

142 

143 

144@subsidiary_router.get("/{sub_id}", responses={404: {"model": ErrorRepr}}) 

145async def get_sub_by_id( 

146 _: Annotated[ 

147 SotransOIDCUserModel, 

148 Depends( 

149 idp.get_current_user( 

150 required_role_names=[SotransRole.company_logistician] 

151 ) 

152 ), 

153 ], 

154 sub_id: PydanticObjectIdPath, 

155 params: BaseGetOneQueryParams = Depends(), 

156) -> SubsidiaryDBModel: 

157 return await subsidiary_grabber.get_one(sub_id, params) 

158 

159 

160@subsidiary_router.get("/{sub_id}/clients") 

161async def get_subsidiaries_assigned_clients( 

162 sub_id: PydanticObjectIdPath, 

163 _: Annotated[ 

164 SotransOIDCUserModel, 

165 Depends( 

166 idp.get_current_user( 

167 required_role_names=[SotransRole.company_director] 

168 ) 

169 ), 

170 ], 

171 params: BaseGetListQueryParams = Depends(), 

172) -> GenericGetListResponse[ClientDBModel]: 

173 return await on_get_assigned_clients_full_by_sub(sub_id, params) 

174 

175 

176@subsidiary_router.get("/{sub_id}/orders") 

177async def get_assigned_orders_by_sub_id( 

178 sub_id: PydanticObjectIdPath, 

179 _: Annotated[ 

180 SotransOIDCUserModel, 

181 Depends( 

182 idp.get_current_user( 

183 required_role_names=[SotransRole.company_director] 

184 ) 

185 ), 

186 ], 

187 params: BaseGetListQueryParams = Depends(), 

188) -> GenericGetListResponse[OrderDBModel]: 

189 params.where = update_search_query( 

190 params.where, 

191 {OrderDBModel.assigned.company.subsidiary.id: str(sub_id)}, 

192 ) 

193 return await orders_data_grabber.get_list(params) 

194 

195 

196@subsidiary_router.get("/{sub_id}/employees") 

197async def get_subsidiaries_employees( 

198 sub_id: PydanticObjectIdPath, 

199 _: Annotated[ 

200 SotransOIDCUserModel, 

201 Depends( 

202 idp.get_current_user( 

203 required_role_names=[SotransRole.company_logistician] 

204 ) 

205 ), 

206 ], 

207) -> GenericGetListResponse[SotransUserDBModel]: 

208 sub = await subsidiaries_col.find_single("_id", sub_id) 

209 if sub is None: 

210 raise NotFoundHTTPError("подразделение") 

211 sub_model = SubsidiaryDBModel(**sub) 

212 if sub_model.employees is None: 

213 return GenericGetListResponse(total=0, items=[]) 

214 return GenericGetListResponse( 

215 total=len(sub_model.employees), items=sub_model.employees 

216 )