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

1from typing import Annotated 

2 

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 

15 

16api_keys_router = APIRouter(prefix="/api-keys") 

17 

18 

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 ) 

34 

35 

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("ключ апи") 

47 

48 

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)