mirror of
				https://github.com/brygphilomena/pysimplesat.git
				synced 2025-11-04 08:47:29 +00:00 
			
		
		
		
	Compare commits
	
		
			2 Commits
		
	
	
		
			ff5a913114
			...
			80010f5b91
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 
						 | 
					80010f5b91 | ||
| 
						 | 
					c2b02bbf27 | 
@ -1,6 +1,6 @@
 | 
			
		||||
[project]
 | 
			
		||||
name = "pysimplesat"
 | 
			
		||||
version = "0.1.7"
 | 
			
		||||
version = "0.1.9"
 | 
			
		||||
authors = [
 | 
			
		||||
  { name="Peter Annabel", email="peter.annabel@gmail.com" },
 | 
			
		||||
]
 | 
			
		||||
 | 
			
		||||
@ -77,7 +77,7 @@ class SimpleSatClient(ABC):
 | 
			
		||||
                method,
 | 
			
		||||
                url,
 | 
			
		||||
                headers=headers,
 | 
			
		||||
                data=data,
 | 
			
		||||
                json=data,
 | 
			
		||||
                params=cast(dict[str, Any], params or {}),
 | 
			
		||||
                stream=stream,
 | 
			
		||||
            )
 | 
			
		||||
 | 
			
		||||
@ -8,6 +8,7 @@ if TYPE_CHECKING:
 | 
			
		||||
 | 
			
		||||
    from pysimplesat.clients.base_client import SimpleSatClient
 | 
			
		||||
    from pysimplesat.types import (
 | 
			
		||||
        JSON,
 | 
			
		||||
        RequestData,
 | 
			
		||||
        RequestMethod,
 | 
			
		||||
        RequestParams,
 | 
			
		||||
@ -112,7 +113,7 @@ class SimpleSatEndpoint:
 | 
			
		||||
        self,
 | 
			
		||||
        method: RequestMethod,
 | 
			
		||||
        endpoint: SimpleSatEndpoint | None = None,
 | 
			
		||||
        data: RequestData | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
        params: RequestParams | None = None,
 | 
			
		||||
        headers: dict[str, str] | None = None,
 | 
			
		||||
        stream: bool = False,  # noqa: FBT001, FBT002
 | 
			
		||||
 | 
			
		||||
@ -26,6 +26,7 @@ class AnswersSearchEndpoint(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Answer]:
 | 
			
		||||
        """
 | 
			
		||||
        Performs a POST request against the /answers/search endpoint and returns an initialized PaginatedResponse object.
 | 
			
		||||
@ -41,12 +42,13 @@ class AnswersSearchEndpoint(
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page": page}
 | 
			
		||||
        return PaginatedResponse(
 | 
			
		||||
            super()._make_request("POST", params=params),
 | 
			
		||||
            super()._make_request("POST", data=data, params=params),
 | 
			
		||||
            Answer,
 | 
			
		||||
            self,
 | 
			
		||||
            "answers",
 | 
			
		||||
            page,
 | 
			
		||||
            params,
 | 
			
		||||
            data,
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
    def post(self, data: JSON | None = None, params: SimpleSatRequestParams | None = None) -> Answer:
 | 
			
		||||
 | 
			
		||||
@ -25,6 +25,7 @@ class QuestionsEndpoint(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Question]:
 | 
			
		||||
        """
 | 
			
		||||
        Performs a GET request against the /questions endpoint and returns an initialized PaginatedResponse object.
 | 
			
		||||
@ -40,12 +41,13 @@ class QuestionsEndpoint(
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page": page}
 | 
			
		||||
        return PaginatedResponse(
 | 
			
		||||
            super()._make_request("GET", params=params),
 | 
			
		||||
            super()._make_request("GET", data=data, params=params),
 | 
			
		||||
            Question,
 | 
			
		||||
            self,
 | 
			
		||||
            "questions",
 | 
			
		||||
            page,
 | 
			
		||||
            params,
 | 
			
		||||
            data,
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
    def get(
 | 
			
		||||
 | 
			
		||||
@ -26,6 +26,7 @@ class ResponsesSearchEndpoint(
 | 
			
		||||
        self,
 | 
			
		||||
        page: int,
 | 
			
		||||
        params: SimpleSatRequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[Response]:
 | 
			
		||||
        """
 | 
			
		||||
        Performs a POST request against the /responses/search endpoint and returns an initialized PaginatedResponse object.
 | 
			
		||||
@ -39,14 +40,15 @@ class ResponsesSearchEndpoint(
 | 
			
		||||
        if params:
 | 
			
		||||
            params["page"] = page
 | 
			
		||||
        else:
 | 
			
		||||
            params = {"page[number]": page}
 | 
			
		||||
            params = {"page": page}
 | 
			
		||||
        return PaginatedResponse(
 | 
			
		||||
            super()._make_request("POST", params=params),
 | 
			
		||||
            super()._make_request("POST", data=data, params=params),
 | 
			
		||||
            Response,
 | 
			
		||||
            self,
 | 
			
		||||
            "responses",
 | 
			
		||||
            page,
 | 
			
		||||
            params,
 | 
			
		||||
            data,
 | 
			
		||||
        )
 | 
			
		||||
 | 
			
		||||
    def post(self, data: JSON | None = None, params: SimpleSatRequestParams | None = None) -> Response:
 | 
			
		||||
 | 
			
		||||
@ -32,6 +32,7 @@ class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]):
 | 
			
		||||
        self,
 | 
			
		||||
        page: int | None = 1,
 | 
			
		||||
        params: TRequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ) -> PaginatedResponse[TModel]:
 | 
			
		||||
        pass
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@ -11,7 +11,7 @@ if TYPE_CHECKING:
 | 
			
		||||
    from pydantic import BaseModel
 | 
			
		||||
    from requests import Response
 | 
			
		||||
 | 
			
		||||
    from pysimplesat.types import RequestParams
 | 
			
		||||
    from pysimplesat.types import RequestParams, JSON
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
TModel = TypeVar("TModel", bound="BaseModel")
 | 
			
		||||
@ -43,6 +43,7 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        endpoint: str,
 | 
			
		||||
        page: int,
 | 
			
		||||
        params: RequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ) -> None:
 | 
			
		||||
        """
 | 
			
		||||
        PaginatedResponse is a wrapper class for handling paginated responses from the
 | 
			
		||||
@ -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, params, data)
 | 
			
		||||
 | 
			
		||||
    def _initialize(
 | 
			
		||||
        self,
 | 
			
		||||
@ -67,6 +68,7 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        endpoint: str,
 | 
			
		||||
        page: int,
 | 
			
		||||
        params: RequestParams | None = None,
 | 
			
		||||
        data: JSON | None = None,
 | 
			
		||||
    ):
 | 
			
		||||
        """
 | 
			
		||||
        Initialize the instance variables using the provided response, endpointmodel, and page size.
 | 
			
		||||
@ -74,7 +76,7 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        Args:
 | 
			
		||||
            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
 | 
			
		||||
            endpoint: The endpoint url to extract the apidata
 | 
			
		||||
        """
 | 
			
		||||
        self.response = response
 | 
			
		||||
        self.response_model = response_model
 | 
			
		||||
@ -94,8 +96,9 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
            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
 | 
			
		||||
        self.data = data
 | 
			
		||||
        self.apidata: list[TModel] = [response_model.model_validate(d) for d in response.json().get(endpoint, {})]
 | 
			
		||||
        self.has_apidata = self.apidata and len(self.apidata) > 0
 | 
			
		||||
        self.index = 0
 | 
			
		||||
 | 
			
		||||
    def get_next_page(self) -> PaginatedResponse[TModel]:
 | 
			
		||||
@ -104,13 +107,13 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
 | 
			
		||||
        Returns:
 | 
			
		||||
            PaginatedResponse[TModel]: The updated PaginatedResponse instance
 | 
			
		||||
            with the data from the next page or None if there is no next page.
 | 
			
		||||
            with the apidata from the next page or None if there is no next page.
 | 
			
		||||
        """
 | 
			
		||||
        if not self.has_next_page or not self.next_page:
 | 
			
		||||
            self.has_data = False
 | 
			
		||||
            self.has_apidata = False
 | 
			
		||||
            return self
 | 
			
		||||
 | 
			
		||||
        next_response = self.endpointmodel.paginated(self.next_page, self.params)
 | 
			
		||||
        next_response = self.endpointmodel.paginated(self.next_page, self.params, self.data)
 | 
			
		||||
        self._initialize(
 | 
			
		||||
            next_response.response,
 | 
			
		||||
            next_response.response_model,
 | 
			
		||||
@ -118,6 +121,7 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
            next_response.endpoint,
 | 
			
		||||
            self.next_page,
 | 
			
		||||
            self.params,
 | 
			
		||||
            self.data,
 | 
			
		||||
        )
 | 
			
		||||
        return self
 | 
			
		||||
 | 
			
		||||
@ -127,19 +131,20 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
 | 
			
		||||
        Returns:
 | 
			
		||||
            PaginatedResponse[TModel]: The updated PaginatedResponse instance
 | 
			
		||||
            with the data from the next page or None if there is no next page.
 | 
			
		||||
            with the apidata from the next page or None if there is no next page.
 | 
			
		||||
        """
 | 
			
		||||
        if not self.has_prev_page or not self.prev_page:
 | 
			
		||||
            self.has_data = False
 | 
			
		||||
            self.has_apidata = False
 | 
			
		||||
            return self
 | 
			
		||||
 | 
			
		||||
        prev_response = self.endpointmodel.paginated(self.prev_page, self.params)
 | 
			
		||||
        prev_response = self.endpointmodel.paginated(self.prev_page, self.params, self.data)
 | 
			
		||||
        self._initialize(
 | 
			
		||||
            prev_response.response,
 | 
			
		||||
            prev_response.response_model,
 | 
			
		||||
            prev_response.endpointmodel,
 | 
			
		||||
            self.prev_page,
 | 
			
		||||
            self.params,
 | 
			
		||||
            self.data,
 | 
			
		||||
        )
 | 
			
		||||
        return self
 | 
			
		||||
 | 
			
		||||
@ -150,8 +155,8 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        Yields:
 | 
			
		||||
            TModel: An instance of the model class for each item in the paginated response.
 | 
			
		||||
        """
 | 
			
		||||
        while self.has_data:
 | 
			
		||||
            yield from self.data
 | 
			
		||||
        while self.has_apidata:
 | 
			
		||||
            yield from self.apidata
 | 
			
		||||
            self.get_next_page()
 | 
			
		||||
 | 
			
		||||
    def __iter__(self):
 | 
			
		||||
@ -170,20 +175,20 @@ class PaginatedResponse(Generic[TModel]):
 | 
			
		||||
        Returns:
 | 
			
		||||
            PaginatedResponse[TModel]: The current instance of the PaginatedResponse.
 | 
			
		||||
        """
 | 
			
		||||
        return self.data
 | 
			
		||||
        return self.apidata
 | 
			
		||||
 | 
			
		||||
    def __next__(self):
 | 
			
		||||
        """
 | 
			
		||||
        Implement the iterator protocol by getting the next item in the data.
 | 
			
		||||
        Implement the iterator protocol by getting the next item in the apidata.
 | 
			
		||||
 | 
			
		||||
        Returns:
 | 
			
		||||
            TModel: The next item in the data.
 | 
			
		||||
            TModel: The next item in the apidata.
 | 
			
		||||
 | 
			
		||||
        Raises:
 | 
			
		||||
            StopIteration: If there are no more items in the data.
 | 
			
		||||
            StopIteration: If there are no more items in the apidata.
 | 
			
		||||
        """
 | 
			
		||||
        if self.index < len(self.data):
 | 
			
		||||
            result = self.data[self.index]
 | 
			
		||||
        if self.index < len(self.apidata):
 | 
			
		||||
            result = self.apidata[self.index]
 | 
			
		||||
            self.index += 1
 | 
			
		||||
            return result
 | 
			
		||||
        else:
 | 
			
		||||
 | 
			
		||||
@ -13,9 +13,11 @@ simplesat_api_client = SimpleSatAPIClient(
 | 
			
		||||
 | 
			
		||||
#surveys = simplesat_api_client.surveys.get()
 | 
			
		||||
#print(surveys)
 | 
			
		||||
data = {"start_date": "2015-04-11T17:00:00Z"}
 | 
			
		||||
 | 
			
		||||
page_responses = simplesat_api_client.responses.search.paginated(1)
 | 
			
		||||
page_responses = simplesat_api_client.responses.search.paginated(1, data=data)
 | 
			
		||||
responses = page_responses.all()
 | 
			
		||||
print(responses)
 | 
			
		||||
for response in responses:
 | 
			
		||||
    print(response.id)
 | 
			
		||||
    pass
 | 
			
		||||
#    print(response.id)
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user