Coverage for src / mcp_server_langgraph / api / pagination.py: 79%

1""" 

2Standardized Pagination Models 

3 

4Provides consistent pagination across all list endpoints for production-grade APIs. 

5 

6Usage: 

7 ```python 

8 from mcp_server_langgraph.api.pagination import PaginationParams, PaginatedResponse 

9 

10 @router.get("/items", response_model=PaginatedResponse[Item]) 

11 async def list_items(pagination: PaginationParams = Depends()): 

12 # Use pagination.offset and pagination.limit for queries 

13 items = await db.query().offset(pagination.offset).limit(pagination.limit).all() 

14 total = await db.query().count() 

15 

16 return create_paginated_response( 

17 data=items, 

18 total=total, 

19 page=pagination.page, 

20 page_size=pagination.page_size 

21 ) 

22 ``` 

23""" 

24 

25import math 

26from typing import Generic, TypeVar 

27 

28from pydantic import BaseModel, Field, computed_field, field_validator 

29 

30# Generic type for paginated data 

31T = TypeVar("T") 

32 

33 

34class PaginationParams(BaseModel): 

35 """ 

36 Pagination parameters for list endpoints 

37 

38 Supports both page-based and offset-based pagination: 

39 - Page-based: page + page_size 

40 - Offset-based: offset + limit 

41 

42 Page-based is automatically converted to offset/limit for database queries. 

43 """ 

44 

45 page: int = Field(default=1, ge=1, description="Page number (1-indexed)", examples=[1, 2, 10]) 

46 page_size: int = Field( 

47 default=20, ge=1, le=1000, description="Number of items per page (max: 1000)", examples=[20, 50, 100] 

48 ) 

49 

50 @field_validator("page_size") 

51 @classmethod 

52 def limit_page_size(cls, v: int) -> int: 

53 """Enforce maximum page size to prevent excessive queries""" 

54 if v > 1000: 

55 return 1000 

56 return v 

57 

58 @computed_field # type: ignore[prop-decorator] 

59 @property 

60 def offset(self) -> int: 

61 """Calculate offset from page and page_size (for database queries)""" 

62 return (self.page - 1) * self.page_size 

63 

64 @computed_field # type: ignore[prop-decorator] 

65 @property 

66 def limit(self) -> int: 

67 """Alias for page_size (for database queries)""" 

68 return self.page_size 

69 

70 model_config = { 

71 "json_schema_extra": { 

72 "examples": [ 

73 {"page": 1, "page_size": 20, "offset": 0, "limit": 20}, 

74 {"page": 3, "page_size": 50, "offset": 100, "limit": 50}, 

75 ] 

76 } 

77 } 

78 

79 

80class PaginationMetadata(BaseModel): 

81 """ 

82 Pagination metadata included in responses 

83 

84 Provides information for clients to navigate pages. 

85 """ 

86 

87 total: int = Field(description="Total number of items across all pages", examples=[100, 1000]) 

88 page: int = Field(ge=1, description="Current page number (1-indexed)", examples=[1, 2, 10]) 

89 page_size: int = Field(ge=1, description="Number of items per page", examples=[20, 50, 100]) 

90 total_pages: int = Field(ge=0, description="Total number of pages", examples=[5, 20, 100]) 

91 

92 @computed_field # type: ignore[prop-decorator] 

93 @property 

94 def has_next(self) -> bool: 

95 """Whether there is a next page""" 

96 return self.page < self.total_pages 

97 

98 @computed_field # type: ignore[prop-decorator] 

99 @property 

100 def has_prev(self) -> bool: 

101 """Whether there is a previous page""" 

102 return self.page > 1 

103 

104 @computed_field # type: ignore[prop-decorator] 

105 @property 

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

107 """Next page number (None if on last page)""" 

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

109 

110 @computed_field # type: ignore[prop-decorator] 

111 @property 

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

113 """Previous page number (None if on first page)""" 

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

115 

116 model_config = { 

117 "json_schema_extra": { 

118 "examples": [ 

119 { 

120 "total": 100, 

121 "page": 2, 

122 "page_size": 20, 

123 "total_pages": 5, 

124 "has_next": True, 

125 "has_prev": True, 

126 "next_page": 3, 

127 "prev_page": 1, 

128 } 

129 ] 

130 } 

131 } 

132 

133 

134class PaginatedResponse(BaseModel, Generic[T]): 

135 """ 

136 Standardized paginated response wrapper 

137 

138 Generic type allows type-safe responses for any data type. 

139 

140 Example: 

141 PaginatedResponse[APIKeyResponse] for API keys 

142 PaginatedResponse[UserResponse] for users 

143 """ 

144 

145 data: list[T] = Field(description="Array of items for the current page") 

146 pagination: PaginationMetadata = Field(description="Pagination metadata for navigation") 

147 

148 model_config = { 

149 "json_schema_extra": { 

150 "examples": [ 

151 { 

152 "data": [{"id": "1", "name": "Item 1"}, {"id": "2", "name": "Item 2"}], 

153 "pagination": { 

154 "total": 100, 

155 "page": 1, 

156 "page_size": 20, 

157 "total_pages": 5, 

158 "has_next": True, 

159 "has_prev": False, 

160 "next_page": 2, 

161 "prev_page": None, 

162 }, 

163 } 

164 ] 

165 } 

166 } 

167 

168 

169def create_paginated_response(data: list[T], total: int, page: int, page_size: int) -> PaginatedResponse[T]: 

170 """ 

171 Helper function to create paginated responses 

172 

173 Args: 

174 data: List of items for current page 

175 total: Total number of items across all pages 

176 page: Current page number (1-indexed) 

177 page_size: Number of items per page 

178 

179 Returns: 

180 PaginatedResponse with data and pagination metadata 

181 

182 Example: 

183 ```python 

184 items = await db.query().offset(offset).limit(limit).all() 

185 total = await db.query().count() 

186 

187 return create_paginated_response( 

188 data=items, 

189 total=total, 

190 page=page, 

191 page_size=page_size 

192 ) 

193 ``` 

194 """ 

195 total_pages = math.ceil(total / page_size) if total > 0 else 0 

196 

197 return PaginatedResponse( 

198 data=data, pagination=PaginationMetadata(total=total, page=page, page_size=page_size, total_pages=total_pages) 

199 )