Coverage for api/endpoints/organization/__init__.py: 73%
57 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1from typing import Annotated
3from errors import endpoint_error_handler
4from exceptions import BadParameterHTTPError, INNNotFound
5from fastapi import APIRouter, Body, Depends
6from handlers.authorization.check_role import assert_single_role_access
7from handlers.grabbers.organizations import orgs_grabber
8from handlers.grabbers.resources import (
9 drivers_data_grabber,
10 trailers_data_grabber,
11 trucks_data_grabber,
12)
13from handlers.organizations import (
14 on_create_organization,
15 on_get_org_completion,
16 on_get_organization_resources,
17 on_inn_update,
18 on_update_organization_by_id,
19)
20from handlers.resources import on_get_resources_for_org
21from keycloak import idp
22from mongodb import drivers_col, trailers_col, trucks_col
23from sotrans_models.models._mongo import PydanticObjectIdPath
24from sotrans_models.models.organizations import (
25 OrganizationCreateModel,
26 OrganizationDBModel,
27 OrganizationUpdateModel,
28 UpdateINNPayload,
29)
30from sotrans_models.models.resources.drivers import DriverDBModel
31from sotrans_models.models.resources.trailers import TrailerDBModel
32from sotrans_models.models.resources.trucks import TruckDBModel
33from sotrans_models.models.responses import (
34 ErrorRepr,
35 GenericGetListResponse,
36 ProfileCompletionResponse,
37)
38from sotrans_models.models.roles import SotransRole
39from sotrans_models.models.users import SotransOIDCUserModel
40from starlette import status
41from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams
43organizations_router = APIRouter(
44 prefix="/organizations", tags=["organizations"]
45)
48@endpoint_error_handler
49@organizations_router.post(
50 "",
51 response_model=OrganizationDBModel,
52 status_code=status.HTTP_201_CREATED,
53)
54async def create_organization(
55 user: Annotated[SotransOIDCUserModel, Depends(idp.get_current_user())],
56 item: OrganizationCreateModel = Body(
57 examples=[
58 {
59 "inn": "843643634653",
60 }
61 ]
62 ),
63):
64 try:
65 return await on_create_organization(item, user)
66 except INNNotFound:
67 raise BadParameterHTTPError("ИНН")
70@organizations_router.get(
71 "",
72 response_model=GenericGetListResponse[OrganizationDBModel],
73 response_model_exclude_unset=True,
74)
75@endpoint_error_handler
76async def get_organizations(
77 _: Annotated[
78 SotransOIDCUserModel,
79 Depends(
80 idp.get_current_user(
81 required_role_names=[SotransRole.company_logistician]
82 )
83 ),
84 ],
85 params: BaseGetListQueryParams = Depends(),
86):
87 return await orgs_grabber.get_list(params)
90@organizations_router.get(
91 "/{org_id}",
92 response_model=OrganizationDBModel,
93 response_model_exclude_unset=True,
94 responses={404: {"model": ErrorRepr}},
95)
96@endpoint_error_handler
97async def get_organization(
98 org_id: PydanticObjectIdPath,
99 user: Annotated[SotransOIDCUserModel, Depends(idp.get_current_user())],
100 params: BaseGetOneQueryParams = Depends(),
101):
102 if user.organization_id != str(org_id):
103 assert_single_role_access(user, SotransRole.company_logistician)
104 return await orgs_grabber.get_one(org_id, params)
107@organizations_router.patch("/{org_id}", response_model=OrganizationDBModel)
108async def update_organization(
109 org_id: PydanticObjectIdPath,
110 item: OrganizationUpdateModel,
111 user: Annotated[SotransOIDCUserModel, Depends(idp.get_current_user())],
112):
113 return await on_update_organization_by_id(org_id, item, user)
116@organizations_router.patch(
117 "/{org_id}/inn", response_model=OrganizationDBModel
118)
119async def update_inn(
120 org_id: PydanticObjectIdPath,
121 inn_payload: UpdateINNPayload,
122 user: Annotated[
123 SotransOIDCUserModel,
124 Depends(
125 idp.get_current_user(
126 required_role_names=[SotransRole.carrier_director]
127 )
128 ),
129 ],
130):
131 return await on_inn_update(org_id, inn_payload, user)
134#####################################
135# Resources
136#####################################
139@organizations_router.get("/{org_id}/drivers")
140async def get_organizations_drivers(
141 user: Annotated[
142 SotransOIDCUserModel,
143 Depends(
144 idp.get_current_user(
145 required_role_names=[SotransRole.carrier_logistician]
146 )
147 ),
148 ],
149 org_id: PydanticObjectIdPath,
150 params: BaseGetListQueryParams = Depends(),
151) -> GenericGetListResponse[DriverDBModel]:
152 return await on_get_organization_resources(
153 drivers_data_grabber, user, org_id, params
154 )
157@organizations_router.get("/{org_id}/trucks")
158async def get_organizations_trucks(
159 user: Annotated[
160 SotransOIDCUserModel,
161 Depends(
162 idp.get_current_user(
163 required_role_names=[SotransRole.carrier_logistician]
164 )
165 ),
166 ],
167 org_id: PydanticObjectIdPath,
168 params: BaseGetListQueryParams = Depends(),
169) -> GenericGetListResponse[TruckDBModel]:
170 return await on_get_organization_resources(
171 trucks_data_grabber, user, org_id, params
172 )
175@organizations_router.get("/{org_id}/trailers")
176async def get_organizations_trailers(
177 org_id: PydanticObjectIdPath,
178 user: Annotated[
179 SotransOIDCUserModel,
180 Depends(
181 idp.get_current_user(
182 required_role_names=[SotransRole.carrier_logistician]
183 )
184 ),
185 ],
186 params: BaseGetListQueryParams = Depends(),
187) -> GenericGetListResponse[TrailerDBModel]:
188 return await on_get_organization_resources(
189 trailers_data_grabber, user, org_id, params
190 )
193@organizations_router.get("/{org_id}/completion")
194async def get_org_completion(
195 org_id: PydanticObjectIdPath,
196 _: Annotated[
197 SotransOIDCUserModel,
198 Depends(
199 idp.get_current_user(
200 required_role_names=[SotransRole.carrier_logistician]
201 )
202 ),
203 ],
204) -> ProfileCompletionResponse:
205 return await on_get_org_completion(org_id)