Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/explore/controller_public_recipes.py: 31%

69 statements  

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

1from uuid import UUID 1a

2 

3import orjson 1a

4from fastapi import APIRouter, Depends, HTTPException, Query, Request 1a

5from pydantic import UUID4 1a

6 

7from mealie.routes._base import controller 1a

8from mealie.routes._base.base_controllers import BasePublicHouseholdExploreController 1a

9from mealie.routes.recipe.recipe_crud_routes import JSONBytes 1a

10from mealie.schema.cookbook.cookbook import ReadCookBook 1a

11from mealie.schema.make_dependable import make_dependable 1a

12from mealie.schema.recipe import Recipe 1a

13from mealie.schema.recipe.recipe import RecipeSummary 1a

14from mealie.schema.recipe.recipe_suggestion import RecipeSuggestionQuery, RecipeSuggestionResponse 1a

15from mealie.schema.response.pagination import PaginationBase, PaginationQuery, RecipeSearchQuery 1a

16 

17router = APIRouter(prefix="/recipes") 1a

18 

19 

20@controller(router) 1a

21class PublicRecipesController(BasePublicHouseholdExploreController): 1a

22 @property 1a

23 def cross_household_cookbooks(self): 1a

24 return self.cross_household_repos.cookbooks 

25 

26 @property 1a

27 def cross_household_recipes(self): 1a

28 return self.cross_household_repos.recipes 

29 

30 @router.get("", response_model=PaginationBase[RecipeSummary]) 1a

31 def get_all( 1a

32 self, 

33 request: Request, 

34 q: PaginationQuery = Depends(make_dependable(PaginationQuery)), 

35 search_query: RecipeSearchQuery = Depends(make_dependable(RecipeSearchQuery)), 

36 categories: list[UUID4 | str] | None = Query(None), 

37 tags: list[UUID4 | str] | None = Query(None), 

38 tools: list[UUID4 | str] | None = Query(None), 

39 foods: list[UUID4 | str] | None = Query(None), 

40 households: list[UUID4 | str] | None = Query(None), 

41 ) -> PaginationBase[RecipeSummary]: 

42 cookbook_data: ReadCookBook | None = None 

43 if search_query.cookbook: 

44 COOKBOOK_NOT_FOUND_EXCEPTION = HTTPException(404, "cookbook not found") 

45 if isinstance(search_query.cookbook, UUID): 

46 cb_match_attr = "id" 

47 else: 

48 try: 

49 UUID(search_query.cookbook) 

50 cb_match_attr = "id" 

51 except ValueError: 

52 cb_match_attr = "slug" 

53 cookbook_data = self.cross_household_cookbooks.get_one(search_query.cookbook, cb_match_attr) 

54 

55 if cookbook_data is None or not cookbook_data.public: 

56 raise COOKBOOK_NOT_FOUND_EXCEPTION 

57 household = self.repos.households.get_one(cookbook_data.household_id) 

58 if not household or household.preferences.private_household: 

59 raise COOKBOOK_NOT_FOUND_EXCEPTION 

60 

61 public_filter = "(household.preferences.privateHousehold = FALSE AND settings.public = TRUE)" 

62 if q.query_filter: 

63 q.query_filter = f"({q.query_filter}) AND {public_filter}" 

64 else: 

65 q.query_filter = public_filter 

66 

67 pagination_response = self.cross_household_recipes.page_all( 

68 pagination=q, 

69 cookbook=cookbook_data, 

70 categories=categories, 

71 tags=tags, 

72 tools=tools, 

73 foods=foods, 

74 households=households, 

75 require_all_categories=search_query.require_all_categories, 

76 require_all_tags=search_query.require_all_tags, 

77 require_all_tools=search_query.require_all_tools, 

78 require_all_foods=search_query.require_all_foods, 

79 search=search_query.search, 

80 ) 

81 

82 # merge default pagination with the request's query params 

83 query_params = q.model_dump() | {**request.query_params} 

84 pagination_response.set_pagination_guides( 

85 self.get_explore_url_path(router.url_path_for("get_all")), 

86 {k: v for k, v in query_params.items() if v is not None}, 

87 ) 

88 

89 json_compatible_response = orjson.dumps(pagination_response.model_dump(by_alias=True)) 

90 

91 # Response is returned directly, to avoid validation and improve performance 

92 return JSONBytes(content=json_compatible_response) 

93 

94 @router.get("/suggestions", response_model=RecipeSuggestionResponse) 1a

95 def suggest_recipes( 1a

96 self, 

97 q: RecipeSuggestionQuery = Depends(make_dependable(RecipeSuggestionQuery)), 

98 foods: list[UUID4] | None = Query(None), 

99 tools: list[UUID4] | None = Query(None), 

100 ) -> RecipeSuggestionResponse: 

101 public_filter = "(household.preferences.privateHousehold = FALSE AND settings.public = TRUE)" 

102 if q.query_filter: 

103 q.query_filter = f"({q.query_filter}) AND {public_filter}" 

104 else: 

105 q.query_filter = public_filter 

106 

107 recipes = self.cross_household_recipes.find_suggested_recipes(q, foods, tools) 

108 response = RecipeSuggestionResponse(items=recipes) 

109 json_compatible_response = orjson.dumps(response.model_dump(by_alias=True)) 

110 

111 # Response is returned directly, to avoid validation and improve performance 

112 return JSONBytes(content=json_compatible_response) 

113 

114 @router.get("/{recipe_slug}", response_model=Recipe) 1a

115 def get_recipe(self, recipe_slug: str) -> Recipe: 1a

116 RECIPE_NOT_FOUND_EXCEPTION = HTTPException(404, "recipe not found") 

117 recipe = self.cross_household_recipes.get_one(recipe_slug) 

118 

119 if not recipe or not recipe.settings.public: 

120 raise RECIPE_NOT_FOUND_EXCEPTION 

121 household = self.repos.households.get_one(recipe.household_id) 

122 if not household or household.preferences.private_household: 

123 raise RECIPE_NOT_FOUND_EXCEPTION 

124 

125 return recipe