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

54 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 17:29 +0000

1from functools import cached_property 1a

2 

3from fastapi import APIRouter, Depends, HTTPException, status 1a

4from pydantic import UUID4 1a

5from sqlalchemy import func, select 1a

6 

7from mealie.db.models.users.users import User 1a

8from mealie.schema.household.household import ( 1a

9 HouseholdCreate, 

10 HouseholdInDB, 

11 HouseholdPagination, 

12 UpdateHouseholdAdmin, 

13) 

14from mealie.schema.mapper import mapper 1a

15from mealie.schema.response.pagination import PaginationQuery 1a

16from mealie.schema.response.responses import ErrorResponse 1a

17from mealie.services.household_services.household_service import HouseholdService 1a

18 

19from .._base import BaseAdminController, controller 1a

20from .._base.mixins import HttpRepo 1a

21 

22router = APIRouter(prefix="/households") 1a

23 

24 

25@controller(router) 1a

26class AdminHouseholdManagementRoutes(BaseAdminController): 1a

27 @cached_property 1a

28 def repo(self): 1a

29 if not self.user: 

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

31 

32 return self.repos.households 

33 

34 # ======================================================================= 

35 # CRUD Operations 

36 

37 @property 1a

38 def mixins(self): 1a

39 return HttpRepo[HouseholdCreate, HouseholdInDB, UpdateHouseholdAdmin]( 

40 self.repo, 

41 self.logger, 

42 self.registered_exceptions, 

43 ) 

44 

45 @router.get("", response_model=HouseholdPagination) 1a

46 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a

47 response = self.repo.page_all( 

48 pagination=q, 

49 override=HouseholdInDB, 

50 ) 

51 

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

53 return response 

54 

55 @router.post("", response_model=HouseholdInDB, status_code=status.HTTP_201_CREATED) 1a

56 def create_one(self, data: HouseholdCreate): 1a

57 return HouseholdService.create_household(self.repos, data) 

58 

59 @router.get("/{item_id}", response_model=HouseholdInDB) 1a

60 def get_one(self, item_id: UUID4): 1a

61 return self.mixins.get_one(item_id) 

62 

63 @router.put("/{item_id}", response_model=HouseholdInDB) 1a

64 def update_one(self, item_id: UUID4, data: UpdateHouseholdAdmin): 1a

65 household = self.repo.get_one(item_id) 

66 

67 if data.preferences: 

68 preferences = self.repos.household_preferences.get_one(value=item_id, key="household_id") 

69 preferences = mapper(data.preferences, preferences) 

70 household.preferences = self.repos.household_preferences.update(item_id, preferences) 

71 

72 if data.name not in ["", household.name]: 

73 # only update the household if the name changed, since the name is the only field that can be updated 

74 household.name = data.name 

75 household = self.repo.update(item_id, household) 

76 

77 return household 

78 

79 @router.delete("/{item_id}", response_model=HouseholdInDB) 1a

80 def delete_one(self, item_id: UUID4): 1a

81 item = self.repo.get_one(item_id) 

82 if item: 

83 stmt = select(func.count(User.id)).filter_by(group_id=item.group_id, household_id=item_id) 

84 user_count = self.session.scalar(stmt) 

85 if user_count: 

86 raise HTTPException( 

87 status_code=status.HTTP_400_BAD_REQUEST, 

88 detail=ErrorResponse.respond(message="Cannot delete household with users"), 

89 ) 

90 

91 return self.mixins.delete_one(item_id)