mirror of
https://github.com/brygphilomena/pysimplesat.git
synced 2025-11-04 08:47:29 +00:00
Compare commits
3 Commits
18ba951ea9
...
774dd7091c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
774dd7091c | ||
|
|
0eba968c11 | ||
|
|
3339dbddc8 |
@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "pysimplesat"
|
name = "pysimplesat"
|
||||||
version = "0.1.1"
|
version = "0.1.4"
|
||||||
authors = [
|
authors = [
|
||||||
{ name="Peter Annabel", email="peter.annabel@gmail.com" },
|
{ name="Peter Annabel", email="peter.annabel@gmail.com" },
|
||||||
]
|
]
|
||||||
|
|||||||
@ -1,8 +1,10 @@
|
|||||||
from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint
|
from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint
|
||||||
from pysimplesat.interfaces import (
|
from pysimplesat.interfaces import (
|
||||||
IGettable,
|
IGettable,
|
||||||
|
IPaginateable,
|
||||||
)
|
)
|
||||||
from pysimplesat.models.simplesat import Question
|
from pysimplesat.models.simplesat import Question
|
||||||
|
from pysimplesat.responses.paginated_response import PaginatedResponse
|
||||||
from pysimplesat.types import (
|
from pysimplesat.types import (
|
||||||
JSON,
|
JSON,
|
||||||
SimpleSatRequestParams,
|
SimpleSatRequestParams,
|
||||||
@ -12,10 +14,43 @@ from pysimplesat.types import (
|
|||||||
class QuestionsEndpoint(
|
class QuestionsEndpoint(
|
||||||
SimpleSatEndpoint,
|
SimpleSatEndpoint,
|
||||||
IGettable[Question, SimpleSatRequestParams],
|
IGettable[Question, SimpleSatRequestParams],
|
||||||
|
IPaginateable[Question, SimpleSatRequestParams],
|
||||||
):
|
):
|
||||||
def __init__(self, client, parent_endpoint=None) -> None:
|
def __init__(self, client, parent_endpoint=None) -> None:
|
||||||
SimpleSatEndpoint.__init__(self, client, "questions", parent_endpoint=parent_endpoint)
|
SimpleSatEndpoint.__init__(self, client, "questions", parent_endpoint=parent_endpoint)
|
||||||
IGettable.__init__(self, Question)
|
IGettable.__init__(self, Question)
|
||||||
|
IPaginateable.__init__(self, Question)
|
||||||
|
|
||||||
|
def paginated(
|
||||||
|
self,
|
||||||
|
page: int,
|
||||||
|
limit: int,
|
||||||
|
params: SimpleSatRequestParams | None = None,
|
||||||
|
) -> PaginatedResponse[Question]:
|
||||||
|
"""
|
||||||
|
Performs a GET request against the /questions endpoint and returns an initialized PaginatedResponse object.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
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.
|
||||||
|
Returns:
|
||||||
|
PaginatedResponse[Question]: The initialized PaginatedResponse object.
|
||||||
|
"""
|
||||||
|
if params:
|
||||||
|
params["page[number]"] = page
|
||||||
|
params["page[size]"] = limit
|
||||||
|
else:
|
||||||
|
params = {"page[number]": page, "page[size]": limit}
|
||||||
|
return PaginatedResponse(
|
||||||
|
super()._make_request("GET", params=params),
|
||||||
|
Question,
|
||||||
|
self,
|
||||||
|
"questions",
|
||||||
|
page,
|
||||||
|
limit,
|
||||||
|
params,
|
||||||
|
)
|
||||||
|
|
||||||
def get(
|
def get(
|
||||||
self,
|
self,
|
||||||
|
|||||||
@ -68,6 +68,17 @@ class Survey(SimpleSatModel):
|
|||||||
survey_type: str | None = Field(default=None, alias="SurveyType")
|
survey_type: str | None = Field(default=None, alias="SurveyType")
|
||||||
brand_name: str | None = Field(default=None, alias="BrandName")
|
brand_name: str | None = Field(default=None, alias="BrandName")
|
||||||
|
|
||||||
|
class Question(SimpleSatModel):
|
||||||
|
id: int | None = Field(default=None, alias="Id")
|
||||||
|
survey: dict[str, int | str] | None = Field(default=None, alias="Survey")
|
||||||
|
order: int | None = Field(default=None, alias="Order")
|
||||||
|
metric: str | None = Field(default=None, alias="Metric")
|
||||||
|
text: str | None = Field(default=None, alias="Text")
|
||||||
|
rating_scale: bool | None = Field(default=None, alias="RatingScale")
|
||||||
|
required: bool | None = Field(default=None, alias="Required")
|
||||||
|
choices: list[str] | None = Field(default=None, alias="Choices")
|
||||||
|
rules: list[dict[str, Any]] | None = Field(default=None, alias="Rules")
|
||||||
|
|
||||||
class CustomerBulk(SimpleSatModel):
|
class CustomerBulk(SimpleSatModel):
|
||||||
request_id: str | None = Field(default=None, alias="RequestId")
|
request_id: str | None = Field(default=None, alias="RequestId")
|
||||||
detail: str | None = Field(default=None, alias="Detail")
|
detail: str | None = Field(default=None, alias="Detail")
|
||||||
|
|||||||
@ -86,26 +86,22 @@ class PaginatedResponse(Generic[TModel]):
|
|||||||
self.limit = limit
|
self.limit = limit
|
||||||
# Get page data from the response body
|
# Get page data from the response body
|
||||||
try:
|
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:
|
except:
|
||||||
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('meta.page', {}))
|
pass
|
||||||
self.params = params
|
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.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
|
||||||
self.index = 0
|
self.index = 0
|
||||||
|
|||||||
@ -2,6 +2,7 @@ import re
|
|||||||
import math
|
import math
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
from urllib.parse import parse_qs, urlparse
|
||||||
|
|
||||||
from requests.structures import CaseInsensitiveDict
|
from requests.structures import CaseInsensitiveDict
|
||||||
|
|
||||||
@ -52,64 +53,33 @@ def parse_response_body(
|
|||||||
print(pagination_info)
|
print(pagination_info)
|
||||||
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
|
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
|
||||||
"""
|
"""
|
||||||
if body.get("current_page") is None:
|
#what goes out
|
||||||
return None
|
|
||||||
has_next_page: bool = False
|
has_next_page: bool = False
|
||||||
has_prev_page: bool = False
|
has_prev_page: bool = False
|
||||||
first_page: int | None = None
|
previous_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
|
|
||||||
next_page: int | None = None
|
next_page: int | None = None
|
||||||
next_page_url: str | None = None
|
|
||||||
next_page_token: str | None = None
|
#what comes in
|
||||||
last_page: int | None = None
|
count: int | None = None
|
||||||
|
next: str | None = None
|
||||||
|
previous: str | None = None
|
||||||
|
|
||||||
result = {}
|
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:
|
if body["next"] is not None:
|
||||||
result["first_page"] = body.get("first_page")
|
result["next_page"] = parse_qs(urlparse(body["next"]).query)['page'][0]
|
||||||
|
|
||||||
if body.get("prev_page") is not None:
|
if body["next"] 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:
|
|
||||||
result["has_next_page"] = True
|
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
|
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"):
|
if body["previous"] is not None:
|
||||||
result["has_prev_page"] = body.get("has_prev_page")
|
result["has_prev_page"] = True
|
||||||
elif body.get("current_page") is not None:
|
else:
|
||||||
if body.get("current_page") > 1:
|
result["has_prev_page"] = False
|
||||||
result["has_prev_page"] = True
|
|
||||||
elif body.get("currentPage") is not None:
|
|
||||||
if body.get("currentPage") > 1:
|
|
||||||
result["has_prev_page"] = True
|
|
||||||
|
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
|||||||
@ -14,6 +14,7 @@ simplesat_api_client = SimpleSatAPIClient(
|
|||||||
#surveys = simplesat_api_client.surveys.get()
|
#surveys = simplesat_api_client.surveys.get()
|
||||||
#print(surveys)
|
#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)
|
||||||
print(answers.data)
|
#print(answers.data)
|
||||||
Loading…
x
Reference in New Issue
Block a user