mirror of
				https://github.com/brygphilomena/pysimplesat.git
				synced 2025-11-04 08:47:29 +00:00 
			
		
		
		
	Compare commits
	
		
			3 Commits
		
	
	
		
			774dd7091c
			...
			ff5a913114
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 
						 | 
					ff5a913114 | ||
| 
						 | 
					74bf53aaeb | ||
| 
						 | 
					7f44c5ed43 | 
@ -1,6 +1,6 @@
 | 
			
		||||
[project]
 | 
			
		||||
name = "pysimplesat"
 | 
			
		||||
version = "0.1.4"
 | 
			
		||||
version = "0.1.7"
 | 
			
		||||
authors = [
 | 
			
		||||
  { name="Peter Annabel", email="peter.annabel@gmail.com" },
 | 
			
		||||
]
 | 
			
		||||
 | 
			
		||||
@ -25,7 +25,6 @@ class AnswersSearchEndpoint(
 | 
			
		||||
    def paginated(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        limit: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Answer]:
 | 
			
		||||
        """
 | 
			
		||||
@ -33,23 +32,20 @@ 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[number]"] = page
 | 
			
		||||
            params["page[size]"] = limit
 | 
			
		||||
            params["page"] = page
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page[number]": page, "page[size]": limit}
 | 
			
		||||
            params = {"page": page}
 | 
			
		||||
        return PaginatedResponse(
 | 
			
		||||
            super()._make_request("POST", params=params),
 | 
			
		||||
            Answer,
 | 
			
		||||
            self,
 | 
			
		||||
            "answers",
 | 
			
		||||
            page,
 | 
			
		||||
            limit,
 | 
			
		||||
            params,
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -24,7 +24,6 @@ class QuestionsEndpoint(
 | 
			
		||||
    def paginated(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        limit: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Question]:
 | 
			
		||||
        """
 | 
			
		||||
@ -32,23 +31,20 @@ 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[number]"] = page
 | 
			
		||||
            params["page[size]"] = limit
 | 
			
		||||
            params["page"] = page
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page[number]": page, "page[size]": limit}
 | 
			
		||||
            params = {"page": page}
 | 
			
		||||
        return PaginatedResponse(
 | 
			
		||||
            super()._make_request("GET", params=params),
 | 
			
		||||
            Question,
 | 
			
		||||
            self,
 | 
			
		||||
            "questions",
 | 
			
		||||
            page,
 | 
			
		||||
            limit,
 | 
			
		||||
            params,
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
@ -66,7 +62,6 @@ 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', {}),
 | 
			
		||||
 | 
			
		||||
@ -25,7 +25,6 @@ class ResponsesSearchEndpoint(
 | 
			
		||||
    def paginated(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        limit: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Response]:
 | 
			
		||||
        """
 | 
			
		||||
@ -33,28 +32,23 @@ 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[number]"] = page
 | 
			
		||||
            params["page[size]"] = limit
 | 
			
		||||
            params["page"] = page
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page[number]": page, "page[size]": limit}
 | 
			
		||||
            params = {"page[number]": page}
 | 
			
		||||
        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.
 | 
			
		||||
 | 
			
		||||
@ -45,7 +45,6 @@ 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', {}),
 | 
			
		||||
 | 
			
		||||
@ -30,8 +30,7 @@ class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]):
 | 
			
		||||
    @abstractmethod
 | 
			
		||||
    def paginated(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        page_size: int,
 | 
			
		||||
        page: int | None = 1,
 | 
			
		||||
        params: TRequestParams | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[TModel]:
 | 
			
		||||
        pass
 | 
			
		||||
 | 
			
		||||
@ -53,6 +53,19 @@ 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")
 | 
			
		||||
 | 
			
		||||
@ -42,7 +42,6 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        endpointmodel: IPaginateable,
 | 
			
		||||
        endpoint: str,
 | 
			
		||||
        page: int,
 | 
			
		||||
        limit: int,
 | 
			
		||||
        params: RequestParams | None = None,
 | 
			
		||||
    ) -> None:
 | 
			
		||||
        """
 | 
			
		||||
@ -58,7 +57,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, limit, params)
 | 
			
		||||
        self._initialize(response, response_model, endpointmodel, endpoint, page, params)
 | 
			
		||||
 | 
			
		||||
    def _initialize(
 | 
			
		||||
        self,
 | 
			
		||||
@ -67,7 +66,6 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        endpointmodel: IPaginateable,
 | 
			
		||||
        endpoint: str,
 | 
			
		||||
        page: int,
 | 
			
		||||
        limit: int,
 | 
			
		||||
        params: RequestParams | None = None,
 | 
			
		||||
    ):
 | 
			
		||||
        """
 | 
			
		||||
@ -77,30 +75,24 @@ 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
 | 
			
		||||
        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.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
 | 
			
		||||
        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
 | 
			
		||||
@ -118,14 +110,13 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
            self.has_data = False
 | 
			
		||||
            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(
 | 
			
		||||
            next_response.response,
 | 
			
		||||
            next_response.response_model,
 | 
			
		||||
            next_response.endpointmodel,
 | 
			
		||||
            next_response.endpoint,
 | 
			
		||||
            self.next_page,
 | 
			
		||||
            next_response.limit,
 | 
			
		||||
            self.params,
 | 
			
		||||
        )
 | 
			
		||||
        return self
 | 
			
		||||
@ -142,13 +133,12 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
            self.has_data = False
 | 
			
		||||
            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(
 | 
			
		||||
            prev_response.response,
 | 
			
		||||
            prev_response.response_model,
 | 
			
		||||
            prev_response.endpointmodel,
 | 
			
		||||
            self.prev_page,
 | 
			
		||||
            prev_response.limit,
 | 
			
		||||
            self.params,
 | 
			
		||||
        )
 | 
			
		||||
        return self
 | 
			
		||||
 | 
			
		||||
@ -21,7 +21,6 @@ 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]
 | 
			
		||||
 | 
			
		||||
@ -66,7 +66,10 @@ def parse_response_body(
 | 
			
		||||
 | 
			
		||||
    result = {}
 | 
			
		||||
    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:
 | 
			
		||||
        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()
 | 
			
		||||
#print(surveys)
 | 
			
		||||
 | 
			
		||||
page_answers = simplesat_api_client.answers.search.paginated(1,1000)
 | 
			
		||||
answers = page_answers.all()
 | 
			
		||||
print(answers)
 | 
			
		||||
#print(answers.data)
 | 
			
		||||
page_responses = simplesat_api_client.responses.search.paginated(1)
 | 
			
		||||
responses = page_responses.all()
 | 
			
		||||
print(responses)
 | 
			
		||||
for response in responses:
 | 
			
		||||
    print(response.id)
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user