Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/organizers/controller_categories.py: 76%

61 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 1a

4from pydantic import UUID4, BaseModel, ConfigDict 1a

5 

6from mealie.routes._base import BaseCrudController, controller 1a

7from mealie.routes._base.mixins import HttpRepo 1a

8from mealie.schema import mapper 1a

9from mealie.schema.recipe import CategoryIn, RecipeCategoryResponse 1a

10from mealie.schema.recipe.recipe import RecipeCategory, RecipeCategoryPagination 1a

11from mealie.schema.recipe.recipe_category import CategoryBase, CategoryOut, CategorySave 1a

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

13from mealie.services import urls 1a

14from mealie.services.event_bus_service.event_types import EventCategoryData, EventOperation, EventTypes 1a

15 

16router = APIRouter(prefix="/categories", tags=["Organizer: Categories"]) 1a

17 

18 

19class CategorySummary(BaseModel): 1a

20 id: UUID4 1a

21 slug: str 1a

22 name: str 1a

23 model_config = ConfigDict(from_attributes=True) 1a

24 

25 

26@controller(router) 1a

27class RecipeCategoryController(BaseCrudController): 1a

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

29 # CRUD Operations 

30 @cached_property 1a

31 def repo(self): 1a

32 return self.repos.categories 1dcb

33 

34 @cached_property 1a

35 def mixins(self): 1a

36 return HttpRepo[CategorySave, CategoryOut, CategorySave](self.repo, self.logger) 1cb

37 

38 @router.get("", response_model=RecipeCategoryPagination) 1a

39 def get_all(self, q: PaginationQuery = Depends(PaginationQuery), search: str | None = None): 1a

40 """Returns a list of available categories in the database""" 

41 response = self.repo.page_all( 1db

42 pagination=q, 

43 override=RecipeCategory, 

44 search=search, 

45 ) 

46 

47 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1db

48 return response 1db

49 

50 @router.post("", status_code=201) 1a

51 def create_one(self, category: CategoryIn): 1a

52 """Creates a Category in the database""" 

53 save_data = mapper.cast(category, CategorySave, group_id=self.group_id) 1cb

54 new_category = self.mixins.create_one(save_data) 1cb

55 if new_category: 55 ↛ 68line 55 didn't jump to line 68 because the condition on line 55 was always true1cb

56 self.publish_event( 1cb

57 event_type=EventTypes.category_created, 

58 document_data=EventCategoryData(operation=EventOperation.create, category_id=new_category.id), 

59 group_id=new_category.group_id, 

60 household_id=None, 

61 message=self.t( 

62 "notifications.generic-created-with-url", 

63 name=new_category.name, 

64 url=urls.category_url(new_category.slug, self.settings.BASE_URL), 

65 ), 

66 ) 

67 

68 return new_category 1cb

69 

70 @router.get("/empty", response_model=list[CategoryBase]) 1a

71 def get_all_empty(self): 1a

72 """Returns a list of categories that do not contain any recipes""" 

73 return self.repos.categories.get_empty() 

74 

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

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

77 """Returns a list of recipes associated with the provided category.""" 

78 category_obj = self.mixins.get_one(item_id) 

79 category_obj = CategorySummary.model_validate(category_obj) 

80 return category_obj 

81 

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

83 def update_one(self, item_id: UUID4, update_data: CategoryIn): 1a

84 """Updates an existing Tag in the database""" 

85 save_data = mapper.cast(update_data, CategorySave, group_id=self.group_id) 

86 category = self.mixins.update_one(save_data, item_id) 

87 

88 if category: 

89 self.publish_event( 

90 event_type=EventTypes.category_updated, 

91 document_data=EventCategoryData(operation=EventOperation.update, category_id=category.id), 

92 group_id=category.group_id, 

93 household_id=None, 

94 message=self.t( 

95 "notifications.generic-updated-with-url", 

96 name=category.name, 

97 url=urls.category_url(category.slug, self.settings.BASE_URL), 

98 ), 

99 ) 

100 

101 return category 

102 

103 @router.delete("/{item_id}") 1a

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

105 """ 

106 Removes a recipe category from the database. Deleting a 

107 category does not impact a recipe. The category will be removed 

108 from any recipes that contain it 

109 """ 

110 if category := self.mixins.delete_one(item_id): 

111 self.publish_event( 

112 event_type=EventTypes.category_deleted, 

113 document_data=EventCategoryData(operation=EventOperation.delete, category_id=category.id), 

114 group_id=category.group_id, 

115 household_id=None, 

116 message=self.t("notifications.generic-deleted", name=category.name), 

117 ) 

118 

119 # ========================================================================= 

120 # Read All Operations 

121 

122 @router.get("/slug/{category_slug}") 1a

123 def get_one_by_slug(self, category_slug: str): 1a

124 """Returns a category object with the associated recieps relating to the category""" 

125 category: RecipeCategory = self.mixins.get_one(category_slug, "slug") 

126 return RecipeCategoryResponse.model_construct( 

127 id=category.id, 

128 slug=category.slug, 

129 name=category.name, 

130 recipes=self.repos.recipes.get_by_categories([category]), 

131 )