-
-
Notifications
You must be signed in to change notification settings - Fork 719
/
pagination.py
70 lines (52 loc) · 2.12 KB
/
pagination.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import enum
from typing import Any, Generic, TypeVar
from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
from humps import camelize
from pydantic import BaseModel
from pydantic.generics import GenericModel
from mealie.schema._mealie import MealieModel
DataT = TypeVar("DataT", bound=BaseModel)
class OrderDirection(str, enum.Enum):
asc = "asc"
desc = "desc"
class PaginationQuery(MealieModel):
page: int = 1
per_page: int = 50
order_by: str = "created_at"
order_direction: OrderDirection = OrderDirection.desc
query_filter: str | None = None
class PaginationBase(GenericModel, Generic[DataT]):
page: int = 1
per_page: int = 10
total: int = 0
total_pages: int = 0
items: list[DataT]
next: str | None
previous: str | None
def _set_next(self, route: str, query_params: dict[str, Any]) -> None:
if self.page >= self.total_pages:
self.next = None
return
# combine params with base route
query_params["page"] = self.page + 1
self.next = PaginationBase.merge_query_parameters(route, query_params)
def _set_prev(self, route: str, query_params: dict[str, Any]) -> None:
if self.page <= 1:
self.previous = None
return
# combine params with base route
query_params["page"] = self.page - 1
self.previous = PaginationBase.merge_query_parameters(route, query_params)
def set_pagination_guides(self, route: str, query_params: dict[str, Any] | None) -> None:
valid_dict: dict[str, Any] = camelize(query_params) if query_params else {}
# sanitize user input
self.page = max(self.page, 1)
self._set_next(route, valid_dict)
self._set_prev(route, valid_dict)
@staticmethod
def merge_query_parameters(url: str, params: dict[str, Any]):
scheme, netloc, path, query_string, fragment = urlsplit(url)
query_params = parse_qs(query_string)
query_params.update(params)
new_query_string = urlencode(query_params, doseq=True)
return urlunsplit((scheme, netloc, path, new_query_string, fragment))