Coverage for api/endpoints/suggest.py: 62%
58 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1from typing import Annotated
3from fastapi import APIRouter, Depends
4from handlers.misc.clients import on_get_client_suggestion
5from handlers.organizations import on_get_organization_suggestion
6from handlers.resources import on_get_resources_for_org
7from handlers.suggestions import (
8 on_get_sub_suggestions,
9 on_get_users_suggestions,
10)
11from keycloak import idp
12from mongodb import drivers_col, trailers_col, trucks_col
13from pydantic import BaseModel
14from sotrans_models.models._mongo import PydanticObjectIdPath
15from sotrans_models.models.roles import SotransRole
16from sotrans_models.models.users import SotransOIDCUserModel
17from utils.check_carriers_resources import verified_mark
19suggestions_router = APIRouter(prefix="/suggestions", tags=["suggestions"])
22class SuggestionsResponseModel(BaseModel):
23 value: str
24 label: str
27@suggestions_router.get("/employees")
28async def get_employees_suggestions(
29 user: Annotated[
30 SotransOIDCUserModel,
31 Depends(
32 idp.get_current_user(
33 required_role_names=[
34 SotransRole.carrier_director,
35 SotransRole.company_logistician,
36 ]
37 )
38 ),
39 ],
40 search_query: str = "",
41 limit: int = 10,
42 skip: int = 0,
43 minimal_role: SotransRole | None = None,
44) -> list[SuggestionsResponseModel]:
45 if limit > 200:
46 limit = 200
47 entries = await on_get_users_suggestions(
48 user, search_query, limit, skip, minimal_role
49 )
50 return [
51 SuggestionsResponseModel(
52 value=str(entry["id"]),
53 label=f"{entry['surname'] if entry.get('surname') else ''} {entry['name'] if entry.get('name') else ''} {entry['patronymic'] if entry.get('patronymic') else ''}",
54 )
55 for entry in entries
56 ]
59@suggestions_router.get("/trucks")
60async def get_trucks_suggestions(
61 user: Annotated[
62 SotransOIDCUserModel,
63 Depends(
64 idp.get_current_user(
65 required_role_names=[SotransRole.carrier_logistician]
66 )
67 ),
68 ],
69 search_query: str = "",
70 limit: int = 10,
71 skip: int = 0,
72) -> list[SuggestionsResponseModel]:
73 results = await on_get_resources_for_org(
74 trucks_col, limit, skip, user, search_query, for_suggestion=True
75 )
76 return [
77 SuggestionsResponseModel(
78 value=str(res.id), label=f"{res.license_plate}{verified_mark(res)}"
79 )
80 for res in results
81 ]
84@suggestions_router.get("/trailers")
85async def get_trailers_suggestions(
86 user: Annotated[
87 SotransOIDCUserModel,
88 Depends(
89 idp.get_current_user(
90 required_role_names=[SotransRole.carrier_logistician]
91 )
92 ),
93 ],
94 search_query: str = "",
95 limit: int = 10,
96 skip: int = 0,
97) -> list[SuggestionsResponseModel]:
98 results = await on_get_resources_for_org(
99 trailers_col, limit, skip, user, search_query, for_suggestion=True
100 )
101 return [
102 SuggestionsResponseModel(
103 value=str(result.id),
104 label=f"{result.license_plate}{verified_mark(result)}",
105 )
106 for result in results
107 ]
110@suggestions_router.get("/drivers")
111async def get_drivers_suggestions(
112 user: Annotated[
113 SotransOIDCUserModel,
114 Depends(
115 idp.get_current_user(
116 required_role_names=[SotransRole.carrier_logistician]
117 )
118 ),
119 ],
120 search_query: str = "",
121 limit: int = 10,
122 skip: int = 0,
123) -> list[SuggestionsResponseModel]:
124 results = await on_get_resources_for_org(
125 drivers_col, limit, skip, user, search_query, for_suggestion=True
126 )
127 return [
128 SuggestionsResponseModel(
129 value=str(result.id),
130 label=f"{result.surname} {result.name} {result.patronymic or ''}{verified_mark(result)}",
131 )
132 for result in results
133 ]
136@suggestions_router.get("/organizations")
137async def get_organization_suggestions(
138 _: Annotated[
139 SotransOIDCUserModel,
140 Depends(
141 idp.get_current_user(
142 required_role_names=[SotransRole.company_logistician]
143 )
144 ),
145 ],
146 search_query: str = "",
147 limit: int = 10,
148 skip: int = 0,
149) -> list[SuggestionsResponseModel]:
150 results = await on_get_organization_suggestion(search_query, limit, skip)
151 return [
152 SuggestionsResponseModel(
153 value=str(result.id),
154 label=(result.short_name or "Без имени") + verified_mark(result),
155 )
156 for result in results
157 ]
160@suggestions_router.get("/clients")
161async def get_client_suggestions(
162 user: Annotated[
163 SotransOIDCUserModel,
164 Depends(
165 idp.get_current_user(
166 required_role_names=[SotransRole.company_logistician]
167 )
168 ),
169 ],
170 search_query: str = "",
171 assignment_enabled: bool = False,
172 limit: int = 10,
173 skip: int = 0,
174) -> list[SuggestionsResponseModel]:
175 results = await on_get_client_suggestion(
176 search_query, user, assignment_enabled, skip, limit
177 )
178 return [
179 SuggestionsResponseModel(
180 value=str(result.id),
181 label=result.name or result.inn,
182 )
183 for result in results
184 ]
187@suggestions_router.get("/organizations/{org_id}/drivers")
188async def get_org_id_drivers_suggestions(
189 org_id: PydanticObjectIdPath,
190 user: Annotated[
191 SotransOIDCUserModel,
192 Depends(
193 idp.get_current_user(
194 required_role_names=[SotransRole.carrier_logistician]
195 )
196 ),
197 ],
198 search_query: str = "",
199 limit: int = 10,
200 skip: int = 0,
201) -> list[SuggestionsResponseModel]:
202 results = await on_get_resources_for_org(
203 drivers_col,
204 limit,
205 skip,
206 user,
207 search_q=search_query,
208 org_id=org_id,
209 for_suggestion=True,
210 )
211 return [
212 SuggestionsResponseModel(
213 value=str(result.id),
214 label=f"{result.surname} {result.name} {result.patronymic or ''}",
215 )
216 for result in results
217 ]
220@suggestions_router.get("/organizations/{org_id}/trailers")
221async def get_trailers_from_org_id_suggestions(
222 org_id: PydanticObjectIdPath,
223 user: Annotated[
224 SotransOIDCUserModel,
225 Depends(
226 idp.get_current_user(
227 required_role_names=[SotransRole.carrier_logistician]
228 )
229 ),
230 ],
231 search_query: str = "",
232 limit: int = 10,
233 skip: int = 0,
234) -> list[SuggestionsResponseModel]:
235 results = await on_get_resources_for_org(
236 trailers_col,
237 limit,
238 skip,
239 user,
240 search_q=search_query,
241 org_id=org_id,
242 for_suggestion=True,
243 )
244 return [
245 SuggestionsResponseModel(
246 value=str(result.id), label=result.license_plate
247 )
248 for result in results
249 ]
252@suggestions_router.get("/organizations/{org_id}/trucks")
253async def get_trucks_suggestions_by_org_id(
254 org_id: PydanticObjectIdPath,
255 user: Annotated[
256 SotransOIDCUserModel,
257 Depends(
258 idp.get_current_user(
259 required_role_names=[SotransRole.carrier_logistician]
260 )
261 ),
262 ],
263 search_query: str = "",
264 limit: int = 10,
265 skip: int = 0,
266) -> list[SuggestionsResponseModel]:
267 results = await on_get_resources_for_org(
268 trucks_col,
269 limit,
270 skip,
271 search_q=search_query,
272 user=user,
273 org_id=org_id,
274 for_suggestion=True,
275 )
276 return [
277 SuggestionsResponseModel(
278 value=str(result.id), label=result.license_plate
279 )
280 for result in results
281 ]
284@suggestions_router.get("/subsidiaries")
285async def get_subsidiaries_suggestions(
286 _: Annotated[
287 SotransOIDCUserModel,
288 Depends(
289 idp.get_current_user(
290 required_role_names=[SotransRole.company_logistician]
291 )
292 ),
293 ],
294 search_query: str = "",
295 limit: int = 10,
296 skip: int = 0,
297) -> list[SuggestionsResponseModel]:
298 return await on_get_sub_suggestions(search_query, skip, limit)