Compare commits

..

No commits in common. "ff5a913114bf5645f9d695825811cf73a4cbcaa5" and "774dd7091cfc8faf8e7f07802228aa2a60e1cc4e" have entirely different histories.

11 changed files with 56 additions and 45 deletions

View File

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

View File

@ -25,6 +25,7 @@ class AnswersSearchEndpoint(
def paginated(
self,
page: int,
limit: int,
params: SimpleSatRequestParams | None = None,
) -> PaginatedResponse[Answer]:
"""
@ -32,20 +33,23 @@ class AnswersSearchEndpoint(
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[Answer]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("POST", params=params),
Answer,
self,
"answers",
page,
limit,
params,
)

View File

@ -24,6 +24,7 @@ class QuestionsEndpoint(
def paginated(
self,
page: int,
limit: int,
params: SimpleSatRequestParams | None = None,
) -> PaginatedResponse[Question]:
"""
@ -31,20 +32,23 @@ class QuestionsEndpoint(
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"] = page
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
Question,
self,
"questions",
page,
limit,
params,
)
@ -62,6 +66,7 @@ class QuestionsEndpoint(
Returns:
Question: The parsed response data.
"""
print("get")
return self._parse_many(
Question,
super()._make_request("GET", data=data, params=params).json().get('questions', {}),

View File

@ -25,6 +25,7 @@ class ResponsesSearchEndpoint(
def paginated(
self,
page: int,
limit: int,
params: SimpleSatRequestParams | None = None,
) -> PaginatedResponse[Response]:
"""
@ -32,23 +33,28 @@ class ResponsesSearchEndpoint(
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[Response]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page[number]": page}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("POST", params=params),
Response,
self,
"responses",
page,
limit,
params,
)
#TODO: How do I paginate a post?
def post(self, data: JSON | None = None, params: SimpleSatRequestParams | None = None) -> Response:
"""
Performs a POST request against the /responses/search endpoint.

View File

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

View File

@ -30,7 +30,8 @@ class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]):
@abstractmethod
def paginated(
self,
page: int | None = 1,
page: int,
page_size: int,
params: TRequestParams | None = None,
) -> PaginatedResponse[TModel]:
pass

View File

@ -53,19 +53,6 @@ class TeamMember(SimpleSatModel):
custom_attributes: dict[str, str | int] | None = Field(default=None, alias="CustomAttributes")
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")
tags: list | None = Field(default=None, alias="Tags")
answers: list[dict[str, Any]] | None = Field(default=None, alias="Answers")

View File

@ -42,6 +42,7 @@ class PaginatedResponse(Generic[TModel]):
endpointmodel: IPaginateable,
endpoint: str,
page: int,
limit: int,
params: RequestParams | None = None,
) -> None:
"""
@ -57,7 +58,7 @@ class PaginatedResponse(Generic[TModel]):
expected model type for the response data. This allows for type-safe handling
of model instances throughout the class.
"""
self._initialize(response, response_model, endpointmodel, endpoint, page, params)
self._initialize(response, response_model, endpointmodel, endpoint, page, limit, params)
def _initialize(
self,
@ -66,6 +67,7 @@ class PaginatedResponse(Generic[TModel]):
endpointmodel: IPaginateable,
endpoint: str,
page: int,
limit: int,
params: RequestParams | None = None,
):
"""
@ -75,24 +77,30 @@ class PaginatedResponse(Generic[TModel]):
response: The raw response object from the API.
endpointmodel (SimpleSatEndpoint[TModel]): The endpointmodel associated with the response.
endpoint: The endpoint url to extract the data
limit (int): The number of items per page.
"""
self.response = response
self.response_model = response_model
self.endpointmodel = endpointmodel
self.endpoint = endpoint
self.limit = limit
# Get page data from the response body
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:
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
try:
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:
pass
self.params = params
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
@ -110,13 +118,14 @@ class PaginatedResponse(Generic[TModel]):
self.has_data = False
return self
next_response = self.endpointmodel.paginated(self.next_page, self.params)
next_response = self.endpointmodel.paginated(self.next_page, self.limit, self.params)
self._initialize(
next_response.response,
next_response.response_model,
next_response.endpointmodel,
next_response.endpoint,
self.next_page,
next_response.limit,
self.params,
)
return self
@ -133,12 +142,13 @@ class PaginatedResponse(Generic[TModel]):
self.has_data = False
return self
prev_response = self.endpointmodel.paginated(self.prev_page, self.params)
prev_response = self.endpointmodel.paginated(self.prev_page, self.limit, self.params)
self._initialize(
prev_response.response,
prev_response.response_model,
prev_response.endpointmodel,
self.prev_page,
prev_response.limit,
self.params,
)
return self

View File

@ -21,6 +21,7 @@ class SimpleSatRequestParams(TypedDict):
customFieldConditions: NotRequired[str]
page_token: NotRequired[str]
page: NotRequired[int]
limit: NotRequired[int]
organization_id: NotRequired[int]
platform: NotRequired[str]
status: NotRequired[str]

View File

@ -66,10 +66,7 @@ def parse_response_body(
result = {}
if body["previous"] is not None:
try:
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
except:
result["prev_page"] = 1
result["prev_page"] = parse_qs(urlparse(body["previous"]).query)['page'][0]
if body["next"] is not None:
result["next_page"] = parse_qs(urlparse(body["next"]).query)['page'][0]

View File

@ -14,8 +14,7 @@ simplesat_api_client = SimpleSatAPIClient(
#surveys = simplesat_api_client.surveys.get()
#print(surveys)
page_responses = simplesat_api_client.responses.search.paginated(1)
responses = page_responses.all()
print(responses)
for response in responses:
print(response.id)
page_answers = simplesat_api_client.answers.search.paginated(1,1000)
answers = page_answers.all()
print(answers)
#print(answers.data)