mirror of
https://github.com/brygphilomena/pysimplesat.git
synced 2025-11-04 16:57:30 +00:00
Fixed pagination, removed unused pagesize
This commit is contained in:
parent
7f44c5ed43
commit
74bf53aaeb
@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "pysimplesat"
|
name = "pysimplesat"
|
||||||
version = "0.1.5"
|
version = "0.1.6"
|
||||||
authors = [
|
authors = [
|
||||||
{ name="Peter Annabel", email="peter.annabel@gmail.com" },
|
{ name="Peter Annabel", email="peter.annabel@gmail.com" },
|
||||||
]
|
]
|
||||||
|
|||||||
@ -25,7 +25,6 @@ class AnswersSearchEndpoint(
|
|||||||
def paginated(
|
def paginated(
|
||||||
self,
|
self,
|
||||||
page: int,
|
page: int,
|
||||||
limit: int,
|
|
||||||
params: SimpleSatRequestParams | None = None,
|
params: SimpleSatRequestParams | None = None,
|
||||||
) -> PaginatedResponse[Answer]:
|
) -> PaginatedResponse[Answer]:
|
||||||
"""
|
"""
|
||||||
@ -33,23 +32,20 @@ class AnswersSearchEndpoint(
|
|||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
page (int): The page number to request.
|
page (int): The page number to request.
|
||||||
limit (int): The number of results to return per page.
|
|
||||||
params (dict[str, int | str]): The parameters to send in the request query string.
|
params (dict[str, int | str]): The parameters to send in the request query string.
|
||||||
Returns:
|
Returns:
|
||||||
PaginatedResponse[Answer]: The initialized PaginatedResponse object.
|
PaginatedResponse[Answer]: The initialized PaginatedResponse object.
|
||||||
"""
|
"""
|
||||||
if params:
|
if params:
|
||||||
params["page[number]"] = page
|
params["page"] = page
|
||||||
params["page[size]"] = limit
|
|
||||||
else:
|
else:
|
||||||
params = {"page[number]": page, "page[size]": limit}
|
params = {"page": page}
|
||||||
return PaginatedResponse(
|
return PaginatedResponse(
|
||||||
super()._make_request("POST", params=params),
|
super()._make_request("POST", params=params),
|
||||||
Answer,
|
Answer,
|
||||||
self,
|
self,
|
||||||
"answers",
|
"answers",
|
||||||
page,
|
page,
|
||||||
limit,
|
|
||||||
params,
|
params,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@ -24,7 +24,6 @@ class QuestionsEndpoint(
|
|||||||
def paginated(
|
def paginated(
|
||||||
self,
|
self,
|
||||||
page: int,
|
page: int,
|
||||||
limit: int,
|
|
||||||
params: SimpleSatRequestParams | None = None,
|
params: SimpleSatRequestParams | None = None,
|
||||||
) -> PaginatedResponse[Question]:
|
) -> PaginatedResponse[Question]:
|
||||||
"""
|
"""
|
||||||
@ -32,23 +31,20 @@ class QuestionsEndpoint(
|
|||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
page (int): The page number to request.
|
page (int): The page number to request.
|
||||||
limit (int): The number of results to return per page.
|
|
||||||
params (dict[str, int | str]): The parameters to send in the request query string.
|
params (dict[str, int | str]): The parameters to send in the request query string.
|
||||||
Returns:
|
Returns:
|
||||||
PaginatedResponse[Question]: The initialized PaginatedResponse object.
|
PaginatedResponse[Question]: The initialized PaginatedResponse object.
|
||||||
"""
|
"""
|
||||||
if params:
|
if params:
|
||||||
params["page[number]"] = page
|
params["page"] = page
|
||||||
params["page[size]"] = limit
|
|
||||||
else:
|
else:
|
||||||
params = {"page[number]": page, "page[size]": limit}
|
params = {"page": page}
|
||||||
return PaginatedResponse(
|
return PaginatedResponse(
|
||||||
super()._make_request("GET", params=params),
|
super()._make_request("GET", params=params),
|
||||||
Question,
|
Question,
|
||||||
self,
|
self,
|
||||||
"questions",
|
"questions",
|
||||||
page,
|
page,
|
||||||
limit,
|
|
||||||
params,
|
params,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@ -25,7 +25,6 @@ class ResponsesSearchEndpoint(
|
|||||||
def paginated(
|
def paginated(
|
||||||
self,
|
self,
|
||||||
page: int,
|
page: int,
|
||||||
limit: int,
|
|
||||||
params: SimpleSatRequestParams | None = None,
|
params: SimpleSatRequestParams | None = None,
|
||||||
) -> PaginatedResponse[Response]:
|
) -> PaginatedResponse[Response]:
|
||||||
"""
|
"""
|
||||||
@ -33,28 +32,23 @@ class ResponsesSearchEndpoint(
|
|||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
page (int): The page number to request.
|
page (int): The page number to request.
|
||||||
limit (int): The number of results to return per page.
|
|
||||||
params (dict[str, int | str]): The parameters to send in the request query string.
|
params (dict[str, int | str]): The parameters to send in the request query string.
|
||||||
Returns:
|
Returns:
|
||||||
PaginatedResponse[Response]: The initialized PaginatedResponse object.
|
PaginatedResponse[Response]: The initialized PaginatedResponse object.
|
||||||
"""
|
"""
|
||||||
if params:
|
if params:
|
||||||
params["page[number]"] = page
|
params["page"] = page
|
||||||
params["page[size]"] = limit
|
|
||||||
else:
|
else:
|
||||||
params = {"page[number]": page, "page[size]": limit}
|
params = {"page[number]": page}
|
||||||
return PaginatedResponse(
|
return PaginatedResponse(
|
||||||
super()._make_request("POST", params=params),
|
super()._make_request("POST", params=params),
|
||||||
Response,
|
Response,
|
||||||
self,
|
self,
|
||||||
"responses",
|
"responses",
|
||||||
page,
|
page,
|
||||||
limit,
|
|
||||||
params,
|
params,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
#TODO: How do I paginate a post?
|
|
||||||
def post(self, data: JSON | None = None, params: SimpleSatRequestParams | None = None) -> Response:
|
def post(self, data: JSON | None = None, params: SimpleSatRequestParams | None = None) -> Response:
|
||||||
"""
|
"""
|
||||||
Performs a POST request against the /responses/search endpoint.
|
Performs a POST request against the /responses/search endpoint.
|
||||||
|
|||||||
@ -30,8 +30,7 @@ class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]):
|
|||||||
@abstractmethod
|
@abstractmethod
|
||||||
def paginated(
|
def paginated(
|
||||||
self,
|
self,
|
||||||
page: int,
|
page: int | None = 1,
|
||||||
page_size: int,
|
|
||||||
params: TRequestParams | None = None,
|
params: TRequestParams | None = None,
|
||||||
) -> PaginatedResponse[TModel]:
|
) -> PaginatedResponse[TModel]:
|
||||||
pass
|
pass
|
||||||
|
|||||||
@ -42,7 +42,6 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
endpointmodel: IPaginateable,
|
endpointmodel: IPaginateable,
|
||||||
endpoint: str,
|
endpoint: str,
|
||||||
page: int,
|
page: int,
|
||||||
limit: int,
|
|
||||||
params: RequestParams | None = None,
|
params: RequestParams | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
@ -58,7 +57,7 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
expected model type for the response data. This allows for type-safe handling
|
expected model type for the response data. This allows for type-safe handling
|
||||||
of model instances throughout the class.
|
of model instances throughout the class.
|
||||||
"""
|
"""
|
||||||
self._initialize(response, response_model, endpointmodel, endpoint, page, limit, params)
|
self._initialize(response, response_model, endpointmodel, endpoint, page, params)
|
||||||
|
|
||||||
def _initialize(
|
def _initialize(
|
||||||
self,
|
self,
|
||||||
@ -67,7 +66,6 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
endpointmodel: IPaginateable,
|
endpointmodel: IPaginateable,
|
||||||
endpoint: str,
|
endpoint: str,
|
||||||
page: int,
|
page: int,
|
||||||
limit: int,
|
|
||||||
params: RequestParams | None = None,
|
params: RequestParams | None = None,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
@ -77,30 +75,25 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
response: The raw response object from the API.
|
response: The raw response object from the API.
|
||||||
endpointmodel (SimpleSatEndpoint[TModel]): The endpointmodel associated with the response.
|
endpointmodel (SimpleSatEndpoint[TModel]): The endpointmodel associated with the response.
|
||||||
endpoint: The endpoint url to extract the data
|
endpoint: The endpoint url to extract the data
|
||||||
limit (int): The number of items per page.
|
|
||||||
"""
|
"""
|
||||||
self.response = response
|
self.response = response
|
||||||
self.response_model = response_model
|
self.response_model = response_model
|
||||||
self.endpointmodel = endpointmodel
|
self.endpointmodel = endpointmodel
|
||||||
self.endpoint = endpoint
|
self.endpoint = endpoint
|
||||||
self.limit = limit
|
|
||||||
# Get page data from the response body
|
# Get page data from the response body
|
||||||
try:
|
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')))
|
||||||
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')))
|
if self.parsed_pagination_response is not None:
|
||||||
if self.parsed_pagination_response is not None:
|
# SimpleSat API gives us a handy response to parse for Pagination
|
||||||
# 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_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.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.prev_page: int = self.parsed_pagination_response.get("prev_page", None)
|
self.next_page: int = self.parsed_pagination_response.get("next_page", None)
|
||||||
self.next_page: int = self.parsed_pagination_response.get("next_page", None)
|
else:
|
||||||
else:
|
self.has_next_page: bool = True
|
||||||
# Haven't worked on this yet
|
self.has_prev_page: bool = page > 1
|
||||||
self.has_next_page: bool = True
|
self.prev_page = page - 1 if page > 1 else 1
|
||||||
self.has_prev_page: bool = page > 1
|
self.next_page = page + 1
|
||||||
self.prev_page = page - 1 if page > 1 else 1
|
print(self.parsed_pagination_response)
|
||||||
self.next_page = page + 1
|
|
||||||
except:
|
|
||||||
pass
|
|
||||||
self.params = params
|
self.params = params
|
||||||
self.data: list[TModel] = [response_model.model_validate(d) for d in response.json().get(endpoint, {})]
|
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.has_data = self.data and len(self.data) > 0
|
||||||
@ -118,14 +111,13 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
self.has_data = False
|
self.has_data = False
|
||||||
return self
|
return self
|
||||||
|
|
||||||
next_response = self.endpointmodel.paginated(self.next_page, self.limit, self.params)
|
next_response = self.endpointmodel.paginated(self.next_page, self.params)
|
||||||
self._initialize(
|
self._initialize(
|
||||||
next_response.response,
|
next_response.response,
|
||||||
next_response.response_model,
|
next_response.response_model,
|
||||||
next_response.endpointmodel,
|
next_response.endpointmodel,
|
||||||
next_response.endpoint,
|
next_response.endpoint,
|
||||||
self.next_page,
|
self.next_page,
|
||||||
next_response.limit,
|
|
||||||
self.params,
|
self.params,
|
||||||
)
|
)
|
||||||
return self
|
return self
|
||||||
@ -142,13 +134,12 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
self.has_data = False
|
self.has_data = False
|
||||||
return self
|
return self
|
||||||
|
|
||||||
prev_response = self.endpointmodel.paginated(self.prev_page, self.limit, self.params)
|
prev_response = self.endpointmodel.paginated(self.prev_page, self.params)
|
||||||
self._initialize(
|
self._initialize(
|
||||||
prev_response.response,
|
prev_response.response,
|
||||||
prev_response.response_model,
|
prev_response.response_model,
|
||||||
prev_response.endpointmodel,
|
prev_response.endpointmodel,
|
||||||
self.prev_page,
|
self.prev_page,
|
||||||
prev_response.limit,
|
|
||||||
self.params,
|
self.params,
|
||||||
)
|
)
|
||||||
return self
|
return self
|
||||||
|
|||||||
@ -21,7 +21,6 @@ class SimpleSatRequestParams(TypedDict):
|
|||||||
customFieldConditions: NotRequired[str]
|
customFieldConditions: NotRequired[str]
|
||||||
page_token: NotRequired[str]
|
page_token: NotRequired[str]
|
||||||
page: NotRequired[int]
|
page: NotRequired[int]
|
||||||
limit: NotRequired[int]
|
|
||||||
organization_id: NotRequired[int]
|
organization_id: NotRequired[int]
|
||||||
platform: NotRequired[str]
|
platform: NotRequired[str]
|
||||||
status: NotRequired[str]
|
status: NotRequired[str]
|
||||||
|
|||||||
@ -65,8 +65,12 @@ def parse_response_body(
|
|||||||
previous: str | None = None
|
previous: str | None = None
|
||||||
|
|
||||||
result = {}
|
result = {}
|
||||||
|
print(body)
|
||||||
if body["previous"] is not None:
|
if body["previous"] is not None:
|
||||||
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
|
try:
|
||||||
|
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
|
||||||
|
except:
|
||||||
|
result["prev_page"] = 1
|
||||||
|
|
||||||
if body["next"] is not None:
|
if body["next"] is not None:
|
||||||
result["next_page"] = parse_qs(urlparse(body["next"]).query)['page'][0]
|
result["next_page"] = parse_qs(urlparse(body["next"]).query)['page'][0]
|
||||||
|
|||||||
@ -14,7 +14,8 @@ simplesat_api_client = SimpleSatAPIClient(
|
|||||||
#surveys = simplesat_api_client.surveys.get()
|
#surveys = simplesat_api_client.surveys.get()
|
||||||
#print(surveys)
|
#print(surveys)
|
||||||
|
|
||||||
page_answers = simplesat_api_client.answers.search.paginated(1,1000)
|
page_responses = simplesat_api_client.responses.search.paginated(1)
|
||||||
answers = page_answers.all()
|
responses = page_responses.all()
|
||||||
print(answers)
|
print(responses)
|
||||||
#print(answers.data)
|
for response in responses:
|
||||||
|
print(response.id)
|
||||||
Loading…
x
Reference in New Issue
Block a user