Compare commits

...

6 commits

Author SHA1 Message Date
Peter Annabel
ff5a913114 Remove print statements 2025-08-05 17:51:33 -05:00
Peter Annabel
74bf53aaeb Fixed pagination, removed unused pagesize 2025-08-05 17:18:32 -05:00
Peter Annabel
7f44c5ed43 Fix random print in surveys endpoint
Fix model for Responses
2025-08-05 13:03:01 -05:00
Peter Annabel
774dd7091c Fix questions model 2025-08-05 10:23:59 -05:00
Peter Annabel
0eba968c11 Fixed Questions Endpoint. Added pagination and added missing model 2025-08-05 10:22:07 -05:00
Peter Annabel
3339dbddc8 Fix paginated responses 2025-08-05 10:12:25 -05:00
11 changed files with 92 additions and 90 deletions

View file

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

View file

@ -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,
) )

View file

@ -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,39 @@ 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,
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.
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"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
Question,
self,
"questions",
page,
params,
)
def get( def get(
self, self,
@ -31,7 +62,6 @@ class QuestionsEndpoint(
Returns: Returns:
Question: The parsed response data. Question: The parsed response data.
""" """
print("get")
return self._parse_many( return self._parse_many(
Question, Question,
super()._make_request("GET", data=data, params=params).json().get('questions', {}), super()._make_request("GET", data=data, params=params).json().get('questions', {}),

View file

@ -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.

View file

@ -45,7 +45,6 @@ class SurveysEndpoint(
Returns: Returns:
Survey: The parsed response data. Survey: The parsed response data.
""" """
print("get")
return self._parse_many( return self._parse_many(
Survey, Survey,
super()._make_request("GET", data=data, params=params).json().get('surveys', {}), super()._make_request("GET", data=data, params=params).json().get('surveys', {}),

View file

@ -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

View file

@ -53,6 +53,19 @@ class TeamMember(SimpleSatModel):
custom_attributes: dict[str, str | int] | None = Field(default=None, alias="CustomAttributes") custom_attributes: dict[str, str | int] | None = Field(default=None, alias="CustomAttributes")
class Response(SimpleSatModel): class Response(SimpleSatModel):
id: int | None = Field(default=None, alias="Id")
survey: dict[str, str | int] | None = Field(default=None, alias="Survey")
tags: list[str] | None = Field(default=None, alias="Tags")
created: datetime | None = Field(default=None, alias="Created")
modified: datetime | None = Field(default=None, alias="Modified")
ip_address: str | None = Field(default=None, alias="IPAddress")
ticket: dict[str, Any] | None = Field(default=None, alias="Ticket")
team_members: list[dict[str, Any]] | None = Field(default=None, alias="TeamMembers")
customer: dict[str, Any] | None = Field(default=None, alias="Customer")
answers: list[dict[str, Any]] | None = Field(default=None, alias="Answers")
source: str | None = Field(default=None, alias="Source")
class ResponseCreatePost(SimpleSatModel):
survey_id: int | None = Field(default=None, alias="SurveyId") survey_id: int | None = Field(default=None, alias="SurveyId")
tags: list | None = Field(default=None, alias="Tags") tags: list | None = Field(default=None, alias="Tags")
answers: list[dict[str, Any]] | None = Field(default=None, alias="Answers") answers: list[dict[str, Any]] | None = Field(default=None, alias="Answers")
@ -68,6 +81,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")

View file

@ -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,35 +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')).get('pagination', {}))
except:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('meta.page', {}))
self.params = params
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.first_page: int = self.parsed_pagination_response.get("first_page", None)
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)
self.last_page: int = self.parsed_pagination_response.get("last_page", None)
else: else:
# Haven't worked on this yet
self.has_next_page: bool = True self.has_next_page: bool = True
self.has_prev_page: bool = page > 1 self.has_prev_page: bool = page > 1
self.first_page: int = 1
self.prev_page = page - 1 if page > 1 else 1 self.prev_page = page - 1 if page > 1 else 1
self.next_page = page + 1 self.next_page = page + 1
self.last_page = 999999 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
self.index = 0 self.index = 0
@ -122,14 +110,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
@ -146,13 +133,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

View file

@ -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]

View file

@ -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,36 @@ 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:
try:
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
except:
result["prev_page"] = 1
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")
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 result["has_prev_page"] = True
else:
result["has_prev_page"] = False
return result return result

View file

@ -14,6 +14,8 @@ 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_responses = simplesat_api_client.responses.search.paginated(1)
print(answers) responses = page_responses.all()
print(answers.data) print(responses)
for response in responses:
print(response.id)