Coverage for api/endpoints/admin/api_keys.py: 67%
28 statements
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
« prev ^ index » next coverage.py v7.6.2, created at 2024-10-10 03:02 +0300
1from typing import Annotated
3from exceptions import APIKeyGenerationFailed, NotFoundHTTPError
4from fastapi import APIRouter, Body, Depends, HTTPException
5from handlers.api_keys import on_create_api_key
6from handlers.grabbers.administrative import api_keys_data_grabber
7from keycloak import idp
8from mongodb import api_keys_col
9from sotrans_models.models.admin.api_keys import APIKeyDBModel
10from sotrans_models.models.responses import GenericGetListResponse
11from sotrans_models.models.roles import SotransRole
12from sotrans_models.models.users import SotransOIDCUserModel
13from starlette import status
14from utils.data_grabber import BaseGetListQueryParams
16api_keys_router = APIRouter(prefix="/api-keys")
19@api_keys_router.post("")
20async def create_key(
21 user: Annotated[
22 SotransOIDCUserModel,
23 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])),
24 ],
25 name: str = Body(embed=True),
26) -> APIKeyDBModel:
27 try:
28 return await on_create_api_key(name, user)
29 except APIKeyGenerationFailed:
30 raise HTTPException(
31 status.HTTP_503_SERVICE_UNAVAILABLE,
32 "Генерация уникального ключа не удалась.",
33 )
36@api_keys_router.delete("/{name}", status_code=status.HTTP_204_NO_CONTENT)
37async def remove_key_by_name(
38 name: str,
39 _: Annotated[
40 SotransOIDCUserModel,
41 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])),
42 ],
43):
44 deletion = await api_keys_col.collection.delete_one({"name": name})
45 if deletion.deleted_count == 0:
46 raise NotFoundHTTPError("ключ апи")
49@api_keys_router.get("")
50async def get_all_keys(
51 _: Annotated[
52 SotransOIDCUserModel,
53 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])),
54 ],
55 params: BaseGetListQueryParams = Depends(),
56) -> GenericGetListResponse[APIKeyDBModel]:
57 return await api_keys_data_grabber.get_list(params)