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

57 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 13:45 +0000

1from functools import cached_property 1a

2 

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

4from pydantic import UUID4 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 RecipeTagResponse, TagIn 1a

10from mealie.schema.recipe.recipe import RecipeTag, RecipeTagPagination 1a

11from mealie.schema.recipe.recipe_category import TagSave 1a

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

13from mealie.services import urls 1a

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

15 

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

17 

18 

19@controller(router) 1a

20class TagController(BaseCrudController): 1a

21 @cached_property 1a

22 def repo(self): 1a

23 return self.repos.tags 1bc

24 

25 @cached_property 1a

26 def mixins(self): 1a

27 return HttpRepo(self.repo, self.logger) 1b

28 

29 @router.get("", response_model=RecipeTagPagination) 1a

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

31 """Returns a list of available tags in the database""" 

32 response = self.repo.page_all( 1b

33 pagination=q, 

34 override=RecipeTag, 

35 search=search, 

36 ) 

37 

38 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1b

39 return response 1b

40 

41 @router.get("/empty") 1a

42 def get_empty_tags(self): 1a

43 """Returns a list of tags that do not contain any recipes""" 

44 return self.repo.get_empty() 1b

45 

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

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

48 """Returns a list of recipes associated with the provided tag.""" 

49 return self.mixins.get_one(item_id) 1b

50 

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

52 def create_one(self, tag: TagIn): 1a

53 """Creates a Tag in the database""" 

54 save_data = mapper.cast(tag, TagSave, group_id=self.group_id) 

55 new_tag = self.repo.create(save_data) 

56 

57 if new_tag: 

58 self.publish_event( 

59 event_type=EventTypes.tag_created, 

60 document_data=EventTagData(operation=EventOperation.create, tag_id=new_tag.id), 

61 group_id=new_tag.group_id, 

62 household_id=None, 

63 message=self.t( 

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

65 name=new_tag.name, 

66 url=urls.tag_url(new_tag.slug, self.settings.BASE_URL), 

67 ), 

68 ) 

69 

70 return new_tag 

71 

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

73 def update_one(self, item_id: UUID4, new_tag: TagIn): 1a

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

75 save_data = mapper.cast(new_tag, TagSave, group_id=self.group_id) 

76 tag = self.repo.update(item_id, save_data) 

77 

78 if tag: 

79 self.publish_event( 

80 event_type=EventTypes.tag_updated, 

81 document_data=EventTagData(operation=EventOperation.update, tag_id=tag.id), 

82 group_id=tag.group_id, 

83 household_id=None, 

84 message=self.t( 

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

86 name=tag.name, 

87 url=urls.tag_url(tag.slug, self.settings.BASE_URL), 

88 ), 

89 ) 

90 

91 return tag 

92 

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

94 def delete_recipe_tag(self, item_id: UUID4): 1a

95 """ 

96 Removes a recipe tag from the database. Deleting a 

97 tag does not impact a recipe. The tag will be removed 

98 from any recipes that contain it 

99 """ 

100 

101 try: 1b

102 tag = self.repo.delete(item_id) 1b

103 except Exception as e: 1b

104 raise HTTPException(status.HTTP_400_BAD_REQUEST) from e 1b

105 

106 if tag: 

107 self.publish_event( 

108 event_type=EventTypes.tag_deleted, 

109 document_data=EventTagData(operation=EventOperation.delete, tag_id=tag.id), 

110 group_id=tag.group_id, 

111 household_id=None, 

112 message=self.t("notifications.generic-deleted", name=tag.name), 

113 ) 

114 

115 @router.get("/slug/{tag_slug}", response_model=RecipeTagResponse) 1a

116 async def get_one_by_slug(self, tag_slug: str): 1a

117 return self.repo.get_one(tag_slug, "slug", override_schema=RecipeTagResponse) 1b