Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/response/pagination.py: 97%

1import enum 

2from typing import Annotated, Any 

3from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit 

4 

5from humps import camelize 

6from pydantic import UUID4, BaseModel, Field, field_validator 

7from pydantic_core.core_schema import ValidationInfo 

8 

9from mealie.schema._mealie import MealieModel 

10 

11 

12class OrderDirection(str, enum.Enum): 

13 asc = "asc" 

14 desc = "desc" 

15 

16 

17class OrderByNullPosition(str, enum.Enum): 

18 first = "first" 

19 last = "last" 

20 

21 

22class RecipeSearchQuery(MealieModel): 

23 cookbook: UUID4 | str | None = None 

24 require_all_categories: bool = False 

25 require_all_tags: bool = False 

26 require_all_tools: bool = False 

27 require_all_foods: bool = False 

28 search: str | None = None 

29 _search_seed: str | None = None 

30 

31 

32class RequestQuery(MealieModel): 

33 order_by: str | None = None 

34 order_by_null_position: OrderByNullPosition | None = None 

35 order_direction: OrderDirection = OrderDirection.desc 

36 query_filter: str | None = None 

37 pagination_seed: Annotated[str | None, Field(validate_default=True)] = None 

38 

39 @field_validator("pagination_seed", mode="before") 

40 def validate_randseed(cls, pagination_seed, info: ValidationInfo): 

41 if info.data.get("order_by") == "random" and not pagination_seed: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 raise ValueError("paginationSeed is required when orderBy is random") 

43 return pagination_seed 

44 

45 

46class PaginationQuery(RequestQuery): 

47 page: int = 1 

48 per_page: int = 50 

49 

50 

51class PaginationBase[DataT: BaseModel](BaseModel): 

52 page: int = 1 

53 per_page: int = 10 

54 total: int = 0 

55 total_pages: int = 0 

56 items: list[DataT] 

57 next: str | None = None 

58 previous: str | None = None 

59 

60 def _set_next(self, route: str, query_params: dict[str, Any]) -> None: 

61 if self.page >= self.total_pages: 

62 self.next = None 

63 return 

64 

65 # combine params with base route 

66 query_params["page"] = self.page + 1 

67 self.next = PaginationBase.merge_query_parameters(route, query_params) 

68 

69 def _set_prev(self, route: str, query_params: dict[str, Any]) -> None: 

70 if self.page <= 1: 

71 self.previous = None 

72 return 

73 

74 # combine params with base route 

75 query_params["page"] = self.page - 1 

76 self.previous = PaginationBase.merge_query_parameters(route, query_params) 

77 

78 def set_pagination_guides(self, route: str, query_params: dict[str, Any] | None) -> None: 

79 valid_dict: dict[str, Any] = camelize(query_params) if query_params else {} 

80 

81 # sanitize user input 

82 self.page = max(self.page, 1) 

83 self._set_next(route, valid_dict) 

84 self._set_prev(route, valid_dict) 

85 

86 @staticmethod 

87 def merge_query_parameters(url: str, params: dict[str, Any]): 

88 scheme, netloc, path, query_string, fragment = urlsplit(url) 

89 

90 query_params = parse_qs(query_string) 

91 query_params.update(params) 

92 new_query_string = urlencode(query_params, doseq=True) 

93 

94 return urlunsplit((scheme, netloc, path, new_query_string, fragment))