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

67 statements  

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

1import enum 1a

2from typing import Annotated, Any 1a

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

4 

5from humps import camelize 1a

6from pydantic import UUID4, BaseModel, Field, field_validator 1a

7from pydantic_core.core_schema import ValidationInfo 1a

8 

9from mealie.schema._mealie import MealieModel 1a

10 

11 

12class OrderDirection(str, enum.Enum): 1a

13 asc = "asc" 1a

14 desc = "desc" 1a

15 

16 

17class OrderByNullPosition(str, enum.Enum): 1a

18 first = "first" 1a

19 last = "last" 1a

20 

21 

22class RecipeSearchQuery(MealieModel): 1a

23 cookbook: UUID4 | str | None = None 1a

24 require_all_categories: bool = False 1a

25 require_all_tags: bool = False 1a

26 require_all_tools: bool = False 1a

27 require_all_foods: bool = False 1a

28 search: str | None = None 1a

29 _search_seed: str | None = None 1a

30 

31 

32class RequestQuery(MealieModel): 1a

33 order_by: str | None = None 1a

34 order_by_null_position: OrderByNullPosition | None = None 1a

35 order_direction: OrderDirection = OrderDirection.desc 1a

36 query_filter: str | None = None 1a

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

38 

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

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

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 true1ayzABCOPQRSTUVWXYZ0123bcdDefghiEFGHjklmnopqrsIJ4NLtuvM5Kw6x

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

43 return pagination_seed 1ayzABCOPQRSTUVWXYZ0123bcdDefghiEFGHjklmnopqrsIJ4NLtuvM5Kw6x

44 

45 

46class PaginationQuery(RequestQuery): 1a

47 page: int = 1 1a

48 per_page: int = 50 1a

49 

50 

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

52 page: int = 1 1a

53 per_page: int = 10 1a

54 total: int = 0 1a

55 total_pages: int = 0 1a

56 items: list[DataT] 1a

57 next: str | None = None 1a

58 previous: str | None = None 1a

59 

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

61 if self.page >= self.total_pages: 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

62 self.next = None 1yzABCbcdDefghiEFGHjklmnopqrsIJNtuvKwx

63 return 1yzABCbcdDefghiEFGHjklmnopqrsIJNtuvKwx

64 

65 # combine params with base route 

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

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

68 

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

70 if self.page <= 1: 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

71 self.previous = None 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

72 return 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

73 

74 # combine params with base route 

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

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

77 

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

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

80 

81 # sanitize user input 

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

83 self._set_next(route, valid_dict) 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

84 self._set_prev(route, valid_dict) 1yzABCbcdDefghiEFGHjklmnopqrsIJNLtuvMKwx

85 

86 @staticmethod 1a

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

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

89 

90 query_params = parse_qs(query_string) 1yzABCbcdDefghiEFGHjklmnopqrsIJLtuvMKwx

91 query_params.update(params) 1yzABCbcdDefghiEFGHjklmnopqrsIJLtuvMKwx

92 new_query_string = urlencode(query_params, doseq=True) 1yzABCbcdDefghiEFGHjklmnopqrsIJLtuvMKwx

93 

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