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
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:45 +0000
1from functools import cached_property 1a
3from fastapi import APIRouter, Depends, HTTPException, status 1a
4from pydantic import UUID4 1a
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
16router = APIRouter(prefix="/tags", tags=["Organizer: Tags"]) 1a
19@controller(router) 1a
20class TagController(BaseCrudController): 1a
21 @cached_property 1a
22 def repo(self): 1a
23 return self.repos.tags 1bc
25 @cached_property 1a
26 def mixins(self): 1a
27 return HttpRepo(self.repo, self.logger) 1b
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 )
38 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1b
39 return response 1b
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
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
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)
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 )
70 return new_tag
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)
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 )
91 return tag
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 """
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
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 )
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