Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/admin/admin_management_groups.py: 93%

1from functools import cached_property 

2 

3from fastapi import APIRouter, Depends, HTTPException, status 

4from pydantic import UUID4 

5 

6from mealie.schema.group.group import GroupAdminUpdate 

7from mealie.schema.mapper import mapper 

8from mealie.schema.response.pagination import PaginationQuery 

9from mealie.schema.response.responses import ErrorResponse 

10from mealie.schema.user.user import GroupBase, GroupInDB, GroupPagination 

11from mealie.services.group_services.group_service import GroupService 

12 

13from .._base import BaseAdminController, controller 

14from .._base.mixins import HttpRepo 

15 

16router = APIRouter(prefix="/groups") 

17 

18 

19@controller(router) 

20class AdminGroupManagementRoutes(BaseAdminController): 

21 @cached_property 

22 def repo(self): 

23 if not self.user: 23 ↛ 24line 23 didn't jump to line 24 because the condition on line 23 was never true

24 raise Exception("No user is logged in.") 

25 

26 return self.repos.groups 

27 

28 # ======================================================================= 

29 # CRUD Operations 

30 

31 @property 

32 def mixins(self): 

33 return HttpRepo[GroupBase, GroupInDB, GroupAdminUpdate]( 

34 self.repo, 

35 self.logger, 

36 self.registered_exceptions, 

37 ) 

38 

39 @router.get("", response_model=GroupPagination) 

40 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 

41 response = self.repo.page_all( 

42 pagination=q, 

43 override=GroupInDB, 

44 ) 

45 

46 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

47 return response 

48 

49 @router.post("", response_model=GroupInDB, status_code=status.HTTP_201_CREATED) 

50 def create_one(self, data: GroupBase): 

51 return GroupService.create_group(self.repos, data) 

52 

53 @router.get("/{item_id}", response_model=GroupInDB) 

54 def get_one(self, item_id: UUID4): 

55 return self.mixins.get_one(item_id) 

56 

57 @router.put("/{item_id}", response_model=GroupInDB) 

58 def update_one(self, item_id: UUID4, data: GroupAdminUpdate): 

59 group = self.repo.get_one(item_id) 

60 

61 if data.preferences: 

62 preferences = self.repos.group_preferences.get_one(value=item_id, key="group_id") 

63 preferences = mapper(data.preferences, preferences) 

64 group.preferences = self.repos.group_preferences.update(item_id, preferences) 

65 

66 if data.name not in ["", group.name]: 

67 # only update the group if the name changed, since the name is the only field that can be updated 

68 group.name = data.name 

69 group = self.repo.update(item_id, group) 

70 

71 return group 

72 

73 @router.delete("/{item_id}", response_model=GroupInDB) 

74 def delete_one(self, item_id: UUID4): 

75 item = self.repo.get_one(item_id) 

76 

77 if item and len(item.users) > 0: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 raise HTTPException( 

79 status_code=status.HTTP_400_BAD_REQUEST, 

80 detail=ErrorResponse.respond(message="Cannot delete group with users"), 

81 ) 

82 

83 return self.mixins.delete_one(item_id)