Coverage for api/endpoints/admin/settings.py: 39%

1from fastapi import APIRouter, Depends, HTTPException 

2from keycloak import idp 

3from mongodb import settings_col 

4from pymongo import ReturnDocument 

5from sotrans_models.models.admin.settings import AdminSettings 

6from sotrans_models.models.roles import SotransRole 

7from starlette import status 

8from utils.helper import get_hash 

9 

10settings_router = APIRouter( 

11 prefix="/settings", 

12 dependencies=[ 

13 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])) 

14 ], 

15) 

16 

17 

18@settings_router.patch("") 

19async def put_settings( 

20 settings: AdminSettings, 

21) -> AdminSettings: 

22 existing = await settings_col.collection.find_one({}) 

23 if existing is None: 

24 model_dump = settings.model_dump() 

25 etag = get_hash(model_dump) 

26 return await settings_col.create({**model_dump, **{"etag": etag}}) 

27 etag_q = {"etag": settings.etag} if settings.etag else {} 

28 updated = await settings_col.collection.find_one_and_update( 

29 etag_q, 

30 {"$set": settings.model_dump(exclude_unset=True)}, 

31 return_document=ReturnDocument.AFTER, 

32 ) 

33 if updated is None: 

34 raise HTTPException(status.HTTP_410_GONE, "etag устарел") 

35 etag = get_hash(updated) 

36 await settings_col.collection.update_one( 

37 {"etag": settings.etag}, {"$set": {"etag": etag}} 

38 ) 

39 updated["etag"] = etag 

40 return updated 

41 

42 

43@settings_router.get("") 

44async def get_settings() -> AdminSettings: 

45 settings = await settings_col.collection.find_one({}) 

46 settings_model = AdminSettings(**settings) if settings else AdminSettings() 

47 return settings_model