Datapath/pysimplesat
2
0
Fork 0
mirror of https://github.com/brygphilomena/pysimplesat.git synced 2025-11-04 08:47:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix paginated responses

Browse Source
This commit is contained in:
Peter Annabel 2025-08-05 10:12:25 -05:00
parent 18ba951ea9
commit 3339dbddc8
4 changed files with 36 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[project]
name = "pysimplesat"
version = "0.1.1"
version = "0.1.2"
authors = [
{ name="Peter Annabel", email="peter.annabel@gmail.com" },
]

32
src/pysimplesat/responses/paginated_response.py
View File

@ -86,26 +86,22 @@ class PaginatedResponse(Generic[TModel]):
self.limit = limit
# Get page data from the response body
try:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('pagination', {}))
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')))
if self.parsed_pagination_response is not None:
# SimpleSat API gives us a handy response to parse for Pagination
self.has_next_page: bool = self.parsed_pagination_response.get("has_next_page", False)
self.has_prev_page: bool = self.parsed_pagination_response.get("has_prev_page", False)
self.prev_page: int = self.parsed_pagination_response.get("prev_page", None)
self.next_page: int = self.parsed_pagination_response.get("next_page", None)
else:
# Haven't worked on this yet
self.has_next_page: bool = True
self.has_prev_page: bool = page > 1
self.prev_page = page - 1 if page > 1 else 1
self.next_page = page + 1
except:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('meta.page', {}))
pass
self.params = params
if self.parsed_pagination_response is not None:
# SimpleSat API gives us a handy response to parse for Pagination
self.has_next_page: bool = self.parsed_pagination_response.get("has_next_page", False)
self.has_prev_page: bool = self.parsed_pagination_response.get("has_prev_page", False)
self.first_page: int = self.parsed_pagination_response.get("first_page", None)
self.prev_page: int = self.parsed_pagination_response.get("prev_page", None)
self.next_page: int = self.parsed_pagination_response.get("next_page", None)
self.last_page: int = self.parsed_pagination_response.get("last_page", None)
else:
# Haven't worked on this yet
self.has_next_page: bool = True
self.has_prev_page: bool = page > 1
self.first_page: int = 1
self.prev_page = page - 1 if page > 1 else 1
self.next_page = page + 1
self.last_page = 999999
self.data: list[TModel] = [response_model.model_validate(d) for d in response.json().get(endpoint, {})]
self.has_data = self.data and len(self.data) > 0
self.index = 0

66
src/pysimplesat/utils/helpers.py
View File

@ -2,6 +2,7 @@ import re
import math
from datetime import datetime
from typing import Any
from urllib.parse import parse_qs, urlparse
from requests.structures import CaseInsensitiveDict
@ -52,64 +53,33 @@ def parse_response_body(
print(pagination_info)
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
"""
if body.get("current_page") is None:
return None
#what goes out
has_next_page: bool = False
has_prev_page: bool = False
first_page: int | None = None
prev_page: int | None = None
current_page: int | None = None
current_page_count: int | None = None
limit: int | None = None
total_count: int | None = None
previous_page: int | None = None
next_page: int | None = None
next_page_url: str | None = None
next_page_token: str | None = None
last_page: int | None = None
#what comes in
count: int | None = None
next: str | None = None
previous: str | None = None
result = {}
if body["previous"] is not None:
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
if body.get("first_page") is not None:
result["first_page"] = body.get("first_page")
if body["next"] is not None:
result["next_page"] = parse_qs(urlparse(body["next"]).query)['page'][0]
if body.get("prev_page") is not None:
result["prev_page"] = body.get("prev_page")
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["prev_page"] = body.get("current_page") - 1
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["prev_page"] = body.get("currentPage") - 1
if body.get("next_page") is not None:
result["next_page"] = body.get("next_page")
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["next_page"] = body.get("currentPage") + 1
if body.get("last_page") is not None:
result["last_page"] = body.get("last_page")
elif body.get("lastPage") is not None:
result["last_page"] = body.get("lastPage")
elif body.get("last_page") is None and body.get("current_page") is not None:
result["last_page"] = math.ceil(body.get("total_count")/body.get("limit"))
if body.get("has_next_page"):
result["has_next_page"] = body.get("has_next_page")
elif body.get("current_page") is not None and body.get("next_page") is not None:
if body["next"] is not None:
result["has_next_page"] = True
elif body.get("current_page") is not None and body.get("next_page") is None:
else:
result["has_next_page"] = False
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["has_next_page"] = True
if body.get("has_prev_page"):
result["has_prev_page"] = body.get("has_prev_page")
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["has_prev_page"] = True
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["has_prev_page"] = True
if body["previous"] is not None:
result["has_prev_page"] = True
else:
result["has_prev_page"] = False
return result

5
src/simplesat_scratchpad.py
View File

@ -14,6 +14,7 @@ simplesat_api_client = SimpleSatAPIClient(
#surveys = simplesat_api_client.surveys.get()
#print(surveys)
answers = simplesat_api_client.responses.search.paginated(1,5)
page_answers = simplesat_api_client.answers.search.paginated(1,1000)
answers = page_answers.all()
print(answers)
print(answers.data)
#print(answers.data)