Coverage for src/lexigram/web/pagination/models.py: 59%

59 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 04:37 +0800

1"""Pagination data models for lexigram-web. 

2 

3All of the concrete classes used by the public pagination API live here. The 

4package-level ``__init__`` simply re-exports them for easy imports and keeps 

5its contents free of implementation logic. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass, field 

11from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar 

12 

13from lexigram.contracts.domain.pagination import CursorPage 

14 

15if TYPE_CHECKING: 

16 from collections.abc import Callable 

17 

18T = TypeVar("T") 

19 

20__all__ = [ 

21 "CursorPage", 

22 "Page", 

23 "PageRequest", 

24 "T", 

25 "cursor_page_to_response", 

26 "paginated", 

27] 

28 

29 

30def cursor_page_to_response(page: CursorPage[Any]) -> dict[str, Any]: 

31 """Standard envelope format for cursor pagination.""" 

32 return { 

33 "items": page.items, 

34 "meta": { 

35 "next_cursor": page.next_cursor, 

36 "prev_cursor": page.prev_cursor, 

37 "has_more": page.has_more, 

38 }, 

39 } 

40 

41 

42def paginated( 

43 page_param: str = "page", 

44 size_param: str = "size", 

45 default_page: int = 1, 

46 default_size: int = 20, 

47 max_size: int = 100, 

48) -> Callable[[Callable[..., Any]], Callable[[dict[str, str]], PageRequest]]: 

49 """Create a paginated request class with custom parameters. 

50 

51 The returned callable behaves like the original implementation; it is 

52 merely a convenience helper and is the only piece of logic remaining in 

53 this module. 

54 """ 

55 

56 def decorator(fn: Callable[..., Any]) -> Callable[[dict[str, str]], PageRequest]: 

57 def wrapper(query_params: dict[str, str]) -> PageRequest: 

58 try: 

59 page = int(query_params.get(page_param, default_page)) 

60 page = max(1, page) 

61 except (ValueError, TypeError): 

62 page = default_page 

63 

64 try: 

65 size = int(query_params.get(size_param, default_size)) 

66 size = min(max(1, size), max_size) 

67 except (ValueError, TypeError): 

68 size = default_size 

69 

70 return PageRequest(page=page, size=size) 

71 

72 return wrapper 

73 

74 return decorator 

75 

76 

77@dataclass 

78class PageRequest: 

79 """Standard offset-based pagination request. 

80 

81 This object is independent of any web framework and can be constructed 

82 manually from query parameters. 

83 """ 

84 

85 page: int = 1 

86 size: int = 20 

87 sort_by: str | None = None 

88 sort_order: Literal["asc", "desc"] = "asc" 

89 

90 @property 

91 def offset(self) -> int: 

92 """Calculate offset for database queries.""" 

93 return (self.page - 1) * self.size 

94 

95 @property 

96 def limit(self) -> int: 

97 """Limit for database queries.""" 

98 return self.size 

99 

100 

101@dataclass 

102class Page(Generic[T]): 

103 """Standard offset-based pagination response.""" 

104 

105 items: list[T] 

106 total: int 

107 page: int 

108 size: int 

109 pages: int = field(init=False) 

110 

111 def __post_init__(self) -> None: 

112 self.pages = (self.total + self.size - 1) // self.size if self.size > 0 else 0 

113 

114 @property 

115 def has_next(self) -> bool: 

116 return self.page < self.pages 

117 

118 @property 

119 def has_prev(self) -> bool: 

120 return self.page > 1 

121 

122 @property 

123 def next_page(self) -> int | None: 

124 return self.page + 1 if self.has_next else None 

125 

126 @property 

127 def prev_page(self) -> int | None: 

128 return self.page - 1 if self.has_prev else None 

129 

130 def to_response(self) -> dict[str, Any]: 

131 """Standard envelope format.""" 

132 return { 

133 "items": self.items, 

134 "meta": { 

135 "total": self.total, 

136 "page": self.page, 

137 "size": self.size, 

138 "pages": self.pages, 

139 "has_next": self.has_next, 

140 "has_prev": self.has_prev, 

141 "next_page": self.next_page, 

142 "prev_page": self.prev_page, 

143 }, 

144 }