diff --git a/.gitignore b/.gitignore index b7faf40..eb401b3 100644 --- a/.gitignore +++ b/.gitignore @@ -205,3 +205,4 @@ cython_debug/ marimo/_static/ marimo/_lsp/ __marimo__/ +src/ironscales_scratchpad.py diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..be340bc --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,39 @@ +[project] +name = "pyironscales" +version = "0.1.1" +authors = [ + { name="Peter Annabel", email="peter.annabel@gmail.com" }, +] +description = "A full-featured Python client for the Ironscales API" +readme = "README.md" +requires-python = ">=3.9" +classifiers = [ + "Intended Audience :: Developers", + "Topic :: Software Development :: Libraries :: Python Modules", + "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.10", +] +keywords = [ + "Ironscales", + "API", + "Python", + "Client", + "Annotated", + "Typed", + "MSP", +] +license = "GPL-3.0-only" +license-files = ["LICEN[CS]E*"] +dynamic = ["dependencies"] + +[project.urls] +Homepage = "https://github.com/brygphilomena/pyironscales" +Issues = "https://github.com/brygphilomena/pyironscales/issues" + +[build-system] +requires = ["hatchling >= 1.26", "hatch-requirements-txt"] +build-backend = "hatchling.build" + +[tool.hatch.metadata.hooks.requirements_txt] +files = ["requirements.txt"] \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..04189d9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests==2.32.4 +pydantic==2.11.7 +typing_extensions==4.14.1 diff --git a/src/pyironscales/__init__.py b/src/pyironscales/__init__.py new file mode 100644 index 0000000..bb33544 --- /dev/null +++ b/src/pyironscales/__init__.py @@ -0,0 +1,4 @@ +from pyironscales.clients.ironscales_client import IronscalesAPIClient + +__all__ = ["IronscalesAPIClient"] +__version__ = "0.1.1" diff --git a/src/pyironscales/clients/__init__.py b/src/pyironscales/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/clients/base_client.py b/src/pyironscales/clients/base_client.py new file mode 100644 index 0000000..5bb90cb --- /dev/null +++ b/src/pyironscales/clients/base_client.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import contextlib +import json +import warnings +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Any, cast + +import requests +from requests import Response +from requests.exceptions import Timeout + +from pyironscales.config import Config +from pyironscales.exceptions import ( + AuthenticationFailedException, + ConflictException, + MalformedRequestException, + MethodNotAllowedException, + NotFoundException, + ObjectExistsError, + PermissionsFailedException, + ServerError, + TooManyRequestsException, +) + +if TYPE_CHECKING: + from pyironscales.types import RequestData, RequestMethod, RequestParams + + +class IronscalesClient(ABC): + config: Config = Config() + + @abstractmethod + def _get_headers(self) -> dict[str, str]: + pass + + @abstractmethod + def _get_url(self) -> str: + pass + + def _make_request( + self, + method: RequestMethod, + url: str, + data: RequestData | None = None, + params: RequestParams | None = None, + headers: dict[str, str] | None = None, + retry_count: int = 0, + stream: bool = False, + ) -> Response: + """ + Make an API request using the specified method, endpoint, data, and parameters. + This function isn't intended for use outside of this class. + Please use the available CRUD methods as intended. + + Args: + method (str): The HTTP method to use for the request (e.g., GET, POST, PUT, etc.). + endpoint (str, optional): The endpoint to make the request to. + data (dict, optional): The request data to send. + params (dict, optional): The query parameters to include in the request. + + Returns: + The Response object (see requests.Response). + + Raises: + Exception: If the request returns a status code >= 400. + """ + + if not headers: + headers = self._get_headers() + + # I don't like having to cast the params to a dict, but it's the only way I can get mypy to stop complaining about the type. + # TypedDicts aren't compatible with the dict type and this is the best way I can think of to handle this. + if data: + response = requests.request( + method, + url, + headers=headers, + json=data, + params=cast(dict[str, Any], params or {}), + stream=stream, + ) + else: + response = requests.request( + method, + url, + headers=headers, + params=cast(dict[str, Any], params or {}), + stream=stream, + ) + if not response.ok: + with contextlib.suppress(json.JSONDecodeError): + details: dict = response.json() + if response.status_code == 400: + if details.get("code") == "InvalidObject": + errors = details.get("errors", []) + if len(errors) > 1: + warnings.warn( + "Found multiple errors - we may be masking some important error details. Please submit a Github issue with response.status_code and response.content so we can improve this error handling.", + stacklevel=1, + ) + for error in errors: + if error.get("code") == "ObjectExists": + error.pop("code") # Don't need code in message + raise ObjectExistsError(response, extra_message=json.dumps(error, indent=4)) + + if response.status_code == 400: + raise MalformedRequestException(response) + if response.status_code == 401: + raise AuthenticationFailedException(response) + if response.status_code == 403: + raise PermissionsFailedException(response) + if response.status_code == 404: + raise NotFoundException(response) + if response.status_code == 405: + raise MethodNotAllowedException(response) + if response.status_code == 409: + raise ConflictException(response) + if response.status_code == 429: + raise TooManyRequestsException(response) + if response.status_code == 500: + # if timeout is mentioned anywhere in the response then we'll retry. + # Ideally we'd return immediately on any non-timeout errors (since + # retries won't help much there), but err towards classifying too much + # as retries instead of too little. + if "timeout" in (response.text + response.reason).lower(): + if retry_count < self.config.max_retries: + retry_count += 1 + return self._make_request(method, url, data, params, headers, retry_count) + raise Timeout(response=response) + raise ServerError(response) + + return response diff --git a/src/pyironscales/clients/ironscales_client.py b/src/pyironscales/clients/ironscales_client.py new file mode 100644 index 0000000..b9f64e2 --- /dev/null +++ b/src/pyironscales/clients/ironscales_client.py @@ -0,0 +1,93 @@ +import typing +from datetime import datetime, timedelta, timezone + +from pyironscales.clients.base_client import IronscalesClient +from pyironscales.config import Config + +if typing.TYPE_CHECKING: + from pyironscales.endpoints.ironscales.SurveysEndpoint import SurveysEndpoint + from pyironscales.endpoints.ironscales.AnswersEndpoint import AnswersEndpoint + from pyironscales.endpoints.ironscales.CustomersEndpoint import CustomersEndpoint + from pyironscales.endpoints.ironscales.QuestionsEndpoint import QuestionsEndpoint + from pyironscales.endpoints.ironscales.TeamMembersEndpoint import TeamMembersEndpoint + from pyironscales.endpoints.ironscales.ResponsesEndpoint import ResponsesEndpoint + + +class IronscalesAPIClient(IronscalesClient): + """ + Ironscales API client. Handles the connection to the Ironscales API + and the configuration of all the available endpoints. + """ + + def __init__( + self, + privatekey: str, + scope: str, + ) -> None: + """ + Initializes the client with the given credentials. + + Parameters: + privatekey (str): Your Ironscales API private key. + """ + self.privatekey: str = privatekey + self.scope: list = scope + self.token_expiry_time: datetime = datetime.now(tz=timezone.utc) + + # Grab first access token + self.access_token: str = self._get_access_token() + + # Initializing endpoints + @property + def surveys(self) -> "SurveysEndpoint": + from pyironscales.endpoints.ironscales.SurveysEndpoint import SurveysEndpoint + + return SurveysEndpoint(self) + + def _get_url(self) -> str: + """ + Generates and returns the URL for the Ironscales API endpoints based on the company url and codebase. + Logs in an obtains an access token. + Returns: + str: API URL. + """ + return f"https://appapi.ironscales.com/appapi" + + def _get_access_token(self) -> str: + """ + Performs a request to the ConnectWise Automate API to obtain an access token. + """ + auth_response = self._make_request( + "POST", + f"{self._get_url()}/get-token/", + data={ + "key": self.privatekey, + "scopes": self.scope + }, + headers={ + "Content-Type": "application/json", + "Accept": "application/json" + }, + ) + auth_resp_json = auth_response.json() + token = auth_resp_json["jwt"] + expires_in_sec = auth_resp_json["expires_in"] + self.token_expiry_time = datetime.now(tz=timezone.utc) + timedelta(seconds=expires_in_sec) + return token + + def _refresh_access_token_if_necessary(self): + if datetime.now(tz=timezone.utc) > self.token_expiry_time: + self.access_token = self._get_access_token() + + def _get_headers(self) -> dict[str, str]: + """ + Generates and returns the headers required for making API requests. The access token is refreshed if necessary before returning. + + Returns: + dict[str, str]: Dictionary of headers including Content-Type, Client ID, and Authorization. + """ + return { + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"Bearer {self.access_token}", + } diff --git a/src/pyironscales/config.py b/src/pyironscales/config.py new file mode 100644 index 0000000..e9c2435 --- /dev/null +++ b/src/pyironscales/config.py @@ -0,0 +1,9 @@ +class Config: + def __init__(self, max_retries=3) -> None: + """ + Initializes a new instance of the Config class. + + Args: + max_retries (int): The maximum number of retries for a retryable HTTP operation (500) (default = 3) + """ + self.max_retries = max_retries diff --git a/src/pyironscales/endpoints/__init__.py b/src/pyironscales/endpoints/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/endpoints/base/__init__.py b/src/pyironscales/endpoints/base/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/endpoints/base/base_endpoint.py b/src/pyironscales/endpoints/base/base_endpoint.py new file mode 100644 index 0000000..1ae2200 --- /dev/null +++ b/src/pyironscales/endpoints/base/base_endpoint.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, TypeVar + +if TYPE_CHECKING: + from pydantic import BaseModel + from requests import Response + + from pyironscales.clients.base_client import IronscalesClient + from pyironscales.types import ( + RequestData, + RequestMethod, + RequestParams, + ) + +TChildEndpoint = TypeVar("TChildEndpoint", bound="IronscalesEndpoint") +TModel = TypeVar("TModel", bound="BaseModel") + + +class IronscalesEndpoint: + """ + IronscalesEndpoint is a base class for all Ironscales API endpoint classes. + It provides a generic implementation for interacting with the Ironscales API, + handling requests, parsing responses into model instances, and managing pagination. + + IronscalesEndpoint makes use of a generic type variable TModel, which represents + the expected IronscalesModel type for the endpoint. This allows for type-safe + handling of model instances throughout the class. + + Each derived class should specify the IronscalesModel type it will be working with + when inheriting from IronscalesEndpoint. For example: + class CompanyEndpoint(IronscalesEndpoint[CompanyModel]). + + IronscalesEndpoint provides methods for making API requests and handles pagination + using the PaginatedResponse class. By default, most CRUD methods raise a + NotImplementedError, which should be overridden in derived classes to provide + endpoint-specific implementations. + + IronscalesEndpoint also supports handling nested endpoints, which are referred to as + child endpoints. Child endpoints can be registered and accessed through their parent + endpoint, allowing for easy navigation through related resources in the API. + + Args: + client: The IronscalesAPIClient instance. + endpoint_url (str): The base URL for the specific endpoint. + parent_endpoint (IronscalesEndpoint, optional): The parent endpoint, if applicable. + + Attributes: + client (IronscalesAPIClient): The IronscalesAPIClient instance. + endpoint_url (str): The base URL for the specific endpoint. + _parent_endpoint (IronscalesEndpoint): The parent endpoint, if applicable. + model_parser (ModelParser): An instance of the ModelParser class used for parsing API responses. + _model (Type[TModel]): The model class for the endpoint. + _id (int): The ID of the current resource, if applicable. + _child_endpoints (List[IronscalesEndpoint]): A list of registered child endpoints. + + Generic Type: + TModel: The model class for the endpoint. + """ + + def __init__( + self, + client: IronscalesClient, + endpoint_url: str, + parent_endpoint: IronscalesEndpoint | None = None, + ) -> None: + """ + Initialize a IronscalesEndpoint instance with the client and endpoint base. + + Args: + client: The IronscalesAPIClient instance. + endpoint_base (str): The base URL for the specific endpoint. + """ + self.client = client + self.endpoint_base = endpoint_url + self._parent_endpoint = parent_endpoint + self._id = None + self._child_endpoints: list[IronscalesEndpoint] = [] + + def _register_child_endpoint(self, child_endpoint: TChildEndpoint) -> TChildEndpoint: + """ + Register a child endpoint to the current endpoint. + + Args: + child_endpoint (IronscalesEndpoint): The child endpoint instance. + + Returns: + IronscalesEndpoint: The registered child endpoint. + """ + self._child_endpoints.append(child_endpoint) + return child_endpoint + + def _url_join(self, *args) -> str: # noqa: ANN002 + """ + Join URL parts into a single URL string. + + Args: + *args: The URL parts to join. + + Returns: + str: The joined URL string. + """ + url_parts = [str(arg).strip("/") for arg in args] + return "/".join(url_parts) + + def _get_replaced_url(self) -> str: + if self._id is None: + return self.endpoint_base + return self.endpoint_base.replace("{id}", str(self._id)) + + def _make_request( + self, + method: RequestMethod, + endpoint: IronscalesEndpoint | None = None, + data: RequestData | None = None, + params: RequestParams | None = None, + headers: dict[str, str] | None = None, + stream: bool = False, # noqa: FBT001, FBT002 + ) -> Response: + """ + Make an API request using the specified method, endpoint, data, and parameters. + This function isn't intended for use outside of this class. + Please use the available CRUD methods as intended. + + Args: + method (str): The HTTP method to use for the request (e.g., GET, POST, PUT, etc.). + endpoint (str, optional): The endpoint to make the request to. + data (dict, optional): The request data to send. + params (dict, optional): The query parameters to include in the request. + + Returns: + The Response object (see requests.Response). + + Raises: + Exception: If the request returns a status code >= 400. + """ + url = self._get_endpoint_url() + if endpoint: + url = self._url_join(url, endpoint) + + return self.client._make_request(method, url, data, params, headers, stream) + + def _build_url(self, other_endpoint: IronscalesEndpoint) -> str: + if other_endpoint._parent_endpoint is not None: + parent_url = self._build_url(other_endpoint._parent_endpoint) + if other_endpoint._parent_endpoint._id is not None: + return self._url_join( + parent_url, + other_endpoint._get_replaced_url(), + ) + else: # noqa: RET505 + return self._url_join(parent_url, other_endpoint._get_replaced_url()) + else: + return self._url_join(self.client._get_url(), other_endpoint._get_replaced_url()) + + def _get_endpoint_url(self) -> str: + return self._build_url(self) + + def _parse_many(self, model_type: type[TModel], data: list[dict[str, Any]]) -> list[TModel]: + return [model_type.model_validate(d) for d in data] + + def _parse_one(self, model_type: type[TModel], data: dict[str, Any]) -> TModel: + return model_type.model_validate(data) diff --git a/src/pyironscales/endpoints/ironscales/AnswersEndpoint.py b/src/pyironscales/endpoints/ironscales/AnswersEndpoint.py new file mode 100644 index 0000000..c9dde02 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/AnswersEndpoint.py @@ -0,0 +1,24 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.endpoints.ironscales.AnswersIdEndpoint import AnswersIdEndpoint +from pyironscales.endpoints.ironscales.AnswersSearchEndpoint import AnswersSearchEndpoint + + +class AnswersEndpoint( + IronscalesEndpoint, +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "answers", parent_endpoint=parent_endpoint) + self.search = self._register_child_endpoint(AnswersSearchEndpoint(client, parent_endpoint=self)) + + def id(self, id: int) -> AnswersIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized AnswersIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + AnswersIdEndpoint: The initialized AnswersIdEndpoint object. + """ + child = AnswersIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child diff --git a/src/pyironscales/endpoints/ironscales/AnswersIdEndpoint.py b/src/pyironscales/endpoints/ironscales/AnswersIdEndpoint.py new file mode 100644 index 0000000..2a30981 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/AnswersIdEndpoint.py @@ -0,0 +1,59 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, + IPuttable +) +from pyironscales.models.ironscales import Answer +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class AnswersIdEndpoint( + IronscalesEndpoint, + IGettable[Answer, IronscalesRequestParams], + IPuttable[Answer, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Answer) + IPuttable.__init__(self, Answer) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Answer: + """ + Performs a GET request against the /answers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Answer, + super()._make_request("GET", data=data, params=params).json(), + ) + + def put( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Answer: + """ + Performs a PUT request against the /answers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Answer: The parsed response data. + """ + return self._parse_one( + Answer, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/ironscales/AnswersSearchEndpoint.py b/src/pyironscales/endpoints/ironscales/AnswersSearchEndpoint.py new file mode 100644 index 0000000..7928c83 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/AnswersSearchEndpoint.py @@ -0,0 +1,66 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, + IPaginateable, +) +from pyironscales.models.ironscales import Answer +from pyironscales.responses.paginated_response import PaginatedResponse +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class AnswersSearchEndpoint( + IronscalesEndpoint, + IPostable[Answer, IronscalesRequestParams], + IPaginateable[Answer, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "search", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Answer) + IPaginateable.__init__(self, Answer) + + def paginated( + self, + page: int, + limit: int, + params: IronscalesRequestParams | None = None, + ) -> PaginatedResponse[Answer]: + """ + Performs a POST request against the /answers/search 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[Answer]: 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("POST", params=params), + Answer, + self, + "answers", + page, + limit, + params, + ) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Answer: + """ + Performs a POST request against the /answers/search endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_many(Answer, super()._make_request("POST", data=data, params=params).json().get('answers', {})) diff --git a/src/pyironscales/endpoints/ironscales/CustomersBulkEndpoint.py b/src/pyironscales/endpoints/ironscales/CustomersBulkEndpoint.py new file mode 100644 index 0000000..8be1dab --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/CustomersBulkEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import CustomerBulk +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class CustomersBulkEndpoint( + IronscalesEndpoint, + IPostable[CustomerBulk, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "bulk", parent_endpoint=parent_endpoint) + IPostable.__init__(self, CustomerBulk) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> CustomerBulk: + """ + Performs a POST request against the /customers/bulk endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_one(CustomerBulk, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/ironscales/CustomersEndpoint.py b/src/pyironscales/endpoints/ironscales/CustomersEndpoint.py new file mode 100644 index 0000000..ce17bdf --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/CustomersEndpoint.py @@ -0,0 +1,46 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.endpoints.ironscales.CustomersIdEndpoint import CustomersIdEndpoint +from pyironscales.endpoints.ironscales.CustomersBulkEndpoint import CustomersBulkEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import Customer +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class CustomersEndpoint( + IronscalesEndpoint, + IPostable[Customer, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "customers", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Customer) + self.bulk = self._register_child_endpoint(CustomersBulkEndpoint(client, parent_endpoint=self)) + + def id(self, id: int) -> CustomersIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized CustomersIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + CustomersIdEndpoint: The initialized CustomersIdEndpoint object. + """ + child = CustomersIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Customer: + """ + Performs a POST request against the /customers endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Customer: The parsed response data. + """ + return self._parse_one(Customer, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/ironscales/CustomersIdEndpoint.py b/src/pyironscales/endpoints/ironscales/CustomersIdEndpoint.py new file mode 100644 index 0000000..a777143 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/CustomersIdEndpoint.py @@ -0,0 +1,59 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, + IPuttable +) +from pyironscales.models.ironscales import Customer +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class CustomersIdEndpoint( + IronscalesEndpoint, + IGettable[Customer, IronscalesRequestParams], + IPuttable[Customer, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Customer) + IPuttable.__init__(self, Customer) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Customer: + """ + Performs a GET request against the /customers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Customer, + super()._make_request("GET", data=data, params=params).json(), + ) + + def put( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Customer: + """ + Performs a PUT request against the /customers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Customer: The parsed response data. + """ + return self._parse_one( + Customer, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/ironscales/QuestionsEndpoint.py b/src/pyironscales/endpoints/ironscales/QuestionsEndpoint.py new file mode 100644 index 0000000..a5c3721 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/QuestionsEndpoint.py @@ -0,0 +1,38 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import Question +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class QuestionsEndpoint( + IronscalesEndpoint, + IGettable[Question, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "questions", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Question) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Question: + """ + Performs a GET request against the /questions endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Question: The parsed response data. + """ + print("get") + return self._parse_many( + Question, + super()._make_request("GET", data=data, params=params).json().get('questions', {}), + ) diff --git a/src/pyironscales/endpoints/ironscales/ResponsesCreateOrUpdateEndpoint.py b/src/pyironscales/endpoints/ironscales/ResponsesCreateOrUpdateEndpoint.py new file mode 100644 index 0000000..19a6ee1 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/ResponsesCreateOrUpdateEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class ResponsesCreateOrUpdateEndpoint( + IronscalesEndpoint, + IPostable[Response, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "create-or-update", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Response) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Response: + """ + Performs a POST request against the /responses/create-or-update endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_one(Response, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/ironscales/ResponsesEndpoint.py b/src/pyironscales/endpoints/ironscales/ResponsesEndpoint.py new file mode 100644 index 0000000..165ec16 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/ResponsesEndpoint.py @@ -0,0 +1,26 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.endpoints.ironscales.ResponsesIdEndpoint import ResponsesIdEndpoint +from pyironscales.endpoints.ironscales.ResponsesSearchEndpoint import ResponsesSearchEndpoint +from pyironscales.endpoints.ironscales.ResponsesCreateOrUpdateEndpoint import ResponsesCreateOrUpdateEndpoint + + +class ResponsesEndpoint( + IronscalesEndpoint, +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "responses", parent_endpoint=parent_endpoint) + self.search = self._register_child_endpoint(ResponsesSearchEndpoint(client, parent_endpoint=self)) + self.createorupdate = self._register_child_endpoint(ResponsesCreateOrUpdateEndpoint(client, parent_endpoint=self)) + + def id(self, id: int) -> ResponsesIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized ResponsesIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + ResponsesIdEndpoint: The initialized ResponsesIdEndpoint object. + """ + child = ResponsesIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child diff --git a/src/pyironscales/endpoints/ironscales/ResponsesIdEndpoint.py b/src/pyironscales/endpoints/ironscales/ResponsesIdEndpoint.py new file mode 100644 index 0000000..b3f137a --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/ResponsesIdEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class ResponsesIdEndpoint( + IronscalesEndpoint, + IGettable[Response, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Response) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Response: + """ + Performs a GET request against the /responses/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Response, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/ironscales/ResponsesIdUpdateEndpoint.py b/src/pyironscales/endpoints/ironscales/ResponsesIdUpdateEndpoint.py new file mode 100644 index 0000000..6ec5d72 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/ResponsesIdUpdateEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPuttable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class ResponsesIdUpdateEndpoint( + IronscalesEndpoint, + IPuttable[Response, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IPuttable.__init__(self, Response) + + def put( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Response: + """ + Performs a PUT request against the /responses/{id}/update endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Response: The parsed response data. + """ + return self._parse_one( + Response, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/ironscales/ResponsesSearchEndpoint.py b/src/pyironscales/endpoints/ironscales/ResponsesSearchEndpoint.py new file mode 100644 index 0000000..5611f2f --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/ResponsesSearchEndpoint.py @@ -0,0 +1,68 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.interfaces import ( + IPostable, + IPaginateable, +) +from pyironscales.models.ironscales import Response +from pyironscales.responses.paginated_response import PaginatedResponse +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class ResponsesSearchEndpoint( + Endpoint, + IPostable[Response, RequestParams], + IPaginateable[Response, RequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "search", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Response) + IPaginateable.__init__(self, Response) + + def paginated( + self, + page: int, + limit: int, + params: RequestParams | None = None, + ) -> PaginatedResponse[Response]: + """ + Performs a POST request against the /responses/search 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[Response]: 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("POST", params=params), + Response, + self, + "responses", + page, + limit, + params, + ) + + + #TODO: How do I paginate a post? + def post(self, data: JSON | None = None, params: RequestParams | None = None) -> Response: + """ + Performs a POST request against the /responses/search endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_many(Response, super()._make_request("POST", data=data, params=params).json().get('responses', {})) diff --git a/src/pyironscales/endpoints/ironscales/SurveysEndpoint.py b/src/pyironscales/endpoints/ironscales/SurveysEndpoint.py new file mode 100644 index 0000000..564bb9b --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/SurveysEndpoint.py @@ -0,0 +1,52 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.endpoints.ironscales.SurveysIdEndpoint import SurveysIdEndpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import Survey +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class SurveysEndpoint( + Endpoint, + IGettable[Survey, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "surveys", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Survey) + + def id(self, id: int) -> SurveysIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized SurveysIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + SurveysIdEndpoint: The initialized SurveysIdEndpoint object. + """ + child = SurveysIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def get( + self, + data: JSON | None = None, + params: RequestParams | None = None, + ) -> Survey: + """ + Performs a GET request against the /surveys endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + print("get") + return self._parse_many( + Survey, + super()._make_request("GET", data=data, params=params).json().get('surveys', {}), + ) diff --git a/src/pyironscales/endpoints/ironscales/SurveysIdEmailEndpoint.py b/src/pyironscales/endpoints/ironscales/SurveysIdEmailEndpoint.py new file mode 100644 index 0000000..fbfdcc1 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/SurveysIdEmailEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import SurveyEmail +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class SurveysIdEmailEndpoint( + Endpoint, + IPostable[SurveyEmail, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "email", parent_endpoint=parent_endpoint) + IPostable.__init__(self, SurveyEmail) + + + def post(self, data: JSON | None = None, params: RequestParams | None = None) -> SurveyEmail: + """ + Performs a POST request against the /surveys/{id}/email endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + SurveyEmail: The parsed response data. + """ + return self._parse_one(SurveyEmail, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/ironscales/SurveysIdEndpoint.py b/src/pyironscales/endpoints/ironscales/SurveysIdEndpoint.py new file mode 100644 index 0000000..67c527d --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/SurveysIdEndpoint.py @@ -0,0 +1,10 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.endpoints.ironscales.SurveysIdEmailEndpoint import SurveysIdEmailEndpoint + + +class SurveysIdEndpoint( + Endpoint, +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + self.email = self._register_child_endpoint(SurveysIdEmailEndpoint(client, parent_endpoint=self)) diff --git a/src/pyironscales/endpoints/ironscales/TeamMembersEndpoint.py b/src/pyironscales/endpoints/ironscales/TeamMembersEndpoint.py new file mode 100644 index 0000000..69a5997 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/TeamMembersEndpoint.py @@ -0,0 +1,44 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.endpoints.ironscales.TeamMembersIdEndpoint import TeamMembersIdEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import TeamMember +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class TeamMembersEndpoint( + Endpoint, + IPostable[TeamMember, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "team-members", parent_endpoint=parent_endpoint) + IPostable.__init__(self, TeamMember) + + def id(self, id: int) -> TeamMembersIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized TeamMembersIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + TeamMembersIdEndpoint: The initialized TeamMembersIdEndpoint object. + """ + child = TeamMembersIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def post(self, data: JSON | None = None, params: RequestParams | None = None) -> TeamMember: + """ + Performs a POST request against the /team-members endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + TeamMember: The parsed response data. + """ + return self._parse_one(TeamMember, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/ironscales/TeamMembersIdEndpoint.py b/src/pyironscales/endpoints/ironscales/TeamMembersIdEndpoint.py new file mode 100644 index 0000000..6edf7d6 --- /dev/null +++ b/src/pyironscales/endpoints/ironscales/TeamMembersIdEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import TeamMember +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class TeamMembersIdEndpoint( + Endpoint, + IGettable[TeamMember, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, TeamMember) + + def get( + self, + data: JSON | None = None, + params: RequestParams | None = None, + ) -> TeamMember: + """ + Performs a GET request against the /team-members/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + TeamMember, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/ironscales/__init__.py b/src/pyironscales/endpoints/ironscales/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/endpoints/simplesat/AnswersIdEndpoint.py b/src/pyironscales/endpoints/simplesat/AnswersIdEndpoint.py new file mode 100644 index 0000000..2a30981 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/AnswersIdEndpoint.py @@ -0,0 +1,59 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, + IPuttable +) +from pyironscales.models.ironscales import Answer +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class AnswersIdEndpoint( + IronscalesEndpoint, + IGettable[Answer, IronscalesRequestParams], + IPuttable[Answer, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Answer) + IPuttable.__init__(self, Answer) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Answer: + """ + Performs a GET request against the /answers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Answer, + super()._make_request("GET", data=data, params=params).json(), + ) + + def put( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Answer: + """ + Performs a PUT request against the /answers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Answer: The parsed response data. + """ + return self._parse_one( + Answer, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/simplesat/AnswersSearchEndpoint.py b/src/pyironscales/endpoints/simplesat/AnswersSearchEndpoint.py new file mode 100644 index 0000000..7928c83 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/AnswersSearchEndpoint.py @@ -0,0 +1,66 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, + IPaginateable, +) +from pyironscales.models.ironscales import Answer +from pyironscales.responses.paginated_response import PaginatedResponse +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class AnswersSearchEndpoint( + IronscalesEndpoint, + IPostable[Answer, IronscalesRequestParams], + IPaginateable[Answer, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "search", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Answer) + IPaginateable.__init__(self, Answer) + + def paginated( + self, + page: int, + limit: int, + params: IronscalesRequestParams | None = None, + ) -> PaginatedResponse[Answer]: + """ + Performs a POST request against the /answers/search 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[Answer]: 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("POST", params=params), + Answer, + self, + "answers", + page, + limit, + params, + ) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Answer: + """ + Performs a POST request against the /answers/search endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_many(Answer, super()._make_request("POST", data=data, params=params).json().get('answers', {})) diff --git a/src/pyironscales/endpoints/simplesat/CustomersBulkEndpoint.py b/src/pyironscales/endpoints/simplesat/CustomersBulkEndpoint.py new file mode 100644 index 0000000..8be1dab --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/CustomersBulkEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import CustomerBulk +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class CustomersBulkEndpoint( + IronscalesEndpoint, + IPostable[CustomerBulk, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "bulk", parent_endpoint=parent_endpoint) + IPostable.__init__(self, CustomerBulk) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> CustomerBulk: + """ + Performs a POST request against the /customers/bulk endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_one(CustomerBulk, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/simplesat/CustomersIdEndpoint.py b/src/pyironscales/endpoints/simplesat/CustomersIdEndpoint.py new file mode 100644 index 0000000..a777143 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/CustomersIdEndpoint.py @@ -0,0 +1,59 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, + IPuttable +) +from pyironscales.models.ironscales import Customer +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class CustomersIdEndpoint( + IronscalesEndpoint, + IGettable[Customer, IronscalesRequestParams], + IPuttable[Customer, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Customer) + IPuttable.__init__(self, Customer) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Customer: + """ + Performs a GET request against the /customers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Customer, + super()._make_request("GET", data=data, params=params).json(), + ) + + def put( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Customer: + """ + Performs a PUT request against the /customers/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Customer: The parsed response data. + """ + return self._parse_one( + Customer, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/simplesat/QuestionsEndpoint.py b/src/pyironscales/endpoints/simplesat/QuestionsEndpoint.py new file mode 100644 index 0000000..a5c3721 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/QuestionsEndpoint.py @@ -0,0 +1,38 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import Question +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class QuestionsEndpoint( + IronscalesEndpoint, + IGettable[Question, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "questions", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Question) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> Question: + """ + Performs a GET request against the /questions endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Question: The parsed response data. + """ + print("get") + return self._parse_many( + Question, + super()._make_request("GET", data=data, params=params).json().get('questions', {}), + ) diff --git a/src/pyironscales/endpoints/simplesat/ResponsesCreateOrUpdateEndpoint.py b/src/pyironscales/endpoints/simplesat/ResponsesCreateOrUpdateEndpoint.py new file mode 100644 index 0000000..19a6ee1 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/ResponsesCreateOrUpdateEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class ResponsesCreateOrUpdateEndpoint( + IronscalesEndpoint, + IPostable[Response, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "create-or-update", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Response) + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Response: + """ + Performs a POST request against the /responses/create-or-update endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_one(Response, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/simplesat/ResponsesIdEndpoint.py b/src/pyironscales/endpoints/simplesat/ResponsesIdEndpoint.py new file mode 100644 index 0000000..98aba5e --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/ResponsesIdEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class ResponsesIdEndpoint( + Endpoint, + IGettable[Response, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Response) + + def get( + self, + data: JSON | None = None, + params: RequestParams | None = None, + ) -> Response: + """ + Performs a GET request against the /responses/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + Response, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/simplesat/ResponsesIdUpdateEndpoint.py b/src/pyironscales/endpoints/simplesat/ResponsesIdUpdateEndpoint.py new file mode 100644 index 0000000..c388ee2 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/ResponsesIdUpdateEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import Endpoint +from pyironscales.interfaces import ( + IPuttable, +) +from pyironscales.models.ironscales import Response +from pyironscales.types import ( + JSON, + RequestParams, +) + + +class ResponsesIdUpdateEndpoint( + Endpoint, + IPuttable[Response, RequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + Endpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IPuttable.__init__(self, Response) + + def put( + self, + data: JSON | None = None, + params: RequestParams | None = None, + ) -> Response: + """ + Performs a PUT request against the /responses/{id}/update endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Response: The parsed response data. + """ + return self._parse_one( + Response, + super()._make_request("PUT", data=data, params=params).json(), + ) diff --git a/src/pyironscales/endpoints/simplesat/ResponsesSearchEndpoint.py b/src/pyironscales/endpoints/simplesat/ResponsesSearchEndpoint.py new file mode 100644 index 0000000..ab6483c --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/ResponsesSearchEndpoint.py @@ -0,0 +1,68 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, + IPaginateable, +) +from pyironscales.models.ironscales import Response +from pyironscales.responses.paginated_response import PaginatedResponse +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class ResponsesSearchEndpoint( + IronscalesEndpoint, + IPostable[Response, IronscalesRequestParams], + IPaginateable[Response, IronscalesRequestParams], + +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "search", parent_endpoint=parent_endpoint) + IPostable.__init__(self, Response) + IPaginateable.__init__(self, Response) + + def paginated( + self, + page: int, + limit: int, + params: IronscalesRequestParams | None = None, + ) -> PaginatedResponse[Response]: + """ + Performs a POST request against the /responses/search 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[Response]: 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("POST", params=params), + Response, + self, + "responses", + page, + limit, + params, + ) + + + #TODO: How do I paginate a post? + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> Response: + """ + Performs a POST request against the /responses/search endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + Survey: The parsed response data. + """ + return self._parse_many(Response, super()._make_request("POST", data=data, params=params).json().get('responses', {})) diff --git a/src/pyironscales/endpoints/simplesat/SurveysIdEmailEndpoint.py b/src/pyironscales/endpoints/simplesat/SurveysIdEmailEndpoint.py new file mode 100644 index 0000000..a44d4f7 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/SurveysIdEmailEndpoint.py @@ -0,0 +1,31 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IPostable, +) +from pyironscales.models.ironscales import SurveyEmail +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class SurveysIdEmailEndpoint( + IronscalesEndpoint, + IPostable[SurveyEmail, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "email", parent_endpoint=parent_endpoint) + IPostable.__init__(self, SurveyEmail) + + + def post(self, data: JSON | None = None, params: IronscalesRequestParams | None = None) -> SurveyEmail: + """ + Performs a POST request against the /surveys/{id}/email endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + SurveyEmail: The parsed response data. + """ + return self._parse_one(SurveyEmail, super()._make_request("POST", data=data, params=params).json()) diff --git a/src/pyironscales/endpoints/simplesat/TeamMembersIdEndpoint.py b/src/pyironscales/endpoints/simplesat/TeamMembersIdEndpoint.py new file mode 100644 index 0000000..55a3774 --- /dev/null +++ b/src/pyironscales/endpoints/simplesat/TeamMembersIdEndpoint.py @@ -0,0 +1,37 @@ +from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint +from pyironscales.interfaces import ( + IGettable, +) +from pyironscales.models.ironscales import TeamMember +from pyironscales.types import ( + JSON, + IronscalesRequestParams, +) + + +class TeamMembersIdEndpoint( + IronscalesEndpoint, + IGettable[TeamMember, IronscalesRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, TeamMember) + + def get( + self, + data: JSON | None = None, + params: IronscalesRequestParams | None = None, + ) -> TeamMember: + """ + Performs a GET request against the /team-members/{id} endpoint. + + Parameters: + data (dict[str, Any]): The data to send in the request body. + params (dict[str, int | str]): The parameters to send in the request query string. + Returns: + AuthInformation: The parsed response data. + """ + return self._parse_one( + TeamMember, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pyironscales/exceptions.py b/src/pyironscales/exceptions.py new file mode 100644 index 0000000..2b55cba --- /dev/null +++ b/src/pyironscales/exceptions.py @@ -0,0 +1,89 @@ +import json +from typing import ClassVar +from urllib.parse import urlsplit, urlunsplit + +from requests import JSONDecodeError, Response + + +class IronscalesException(Exception): + _code_explanation: ClassVar[str] = "" # Ex: for 404 "Not Found" + _error_suggestion: ClassVar[str] = "" # Ex: for 404 "Check the URL you are using is correct" + + def __init__(self, req_response: Response, *, extra_message: str = "") -> None: + self.response = req_response + self.extra_message = extra_message + super().__init__(self.message()) + + def _get_sanitized_url(self) -> str: + """ + Simplify URL down to method, hostname, and path. + """ + url_components = urlsplit(self.response.url) + return urlunsplit( + ( + url_components.scheme, + url_components.hostname, + url_components.path, + "", + "", + ) + ) + + def details(self) -> str: + try: + # If response was json, then format it nicely + return json.dumps(self.response.json(), indent=4) + except JSONDecodeError: + return self.response.text + + def message(self) -> str: + return ( + f"A HTTP {self.response.status_code} ({self._code_explanation}) error has occurred while requesting" + f" {self._get_sanitized_url()}.\n{self.response.reason}\n{self._error_suggestion}\n{self.extra_message}" + ).strip() # Remove extra whitespace (Ex: if extra_message == "") + + +class MalformedRequestException(IronscalesException): + _code_explanation = "Bad Request" + _error_suggestion = ( + "The request could not be understood by the server due to malformed syntax. Please check modify your request" + " before retrying." + ) + + +class AuthenticationFailedException(IronscalesException): + _code_explanation = "Unauthorized" + _error_suggestion = "Please check your credentials are correct before retrying." + + +class PermissionsFailedException(IronscalesException): + _code_explanation = "Forbidden" + _error_suggestion = "You may be attempting to access a resource you do not have the appropriate permissions for." + + +class NotFoundException(IronscalesException): + _code_explanation = "Not Found" + _error_suggestion = "You may be attempting to access a resource that has been moved or deleted." + + +class MethodNotAllowedException(IronscalesException): + _code_explanation = "Method Not Allowed" + _error_suggestion = "This resource does not support the HTTP method you are trying to use." + + +class ConflictException(IronscalesException): + _code_explanation = "Conflict" + _error_suggestion = "This resource is possibly in use or conflicts with another record." + +class TooManyRequestsException(IronscalesException): + _code_explanation = "Too Many Requests" + _error_suggestion = "This resource is currently being rate limited. Please wait and try again." + + +class ServerError(IronscalesException): + _code_explanation = "Internal Server Error" + + +class ObjectExistsError(IronscalesException): + _code_explanation = "Object Exists" + _error_suggestion = "This resource already exists." diff --git a/src/pyironscales/interfaces.py b/src/pyironscales/interfaces.py new file mode 100644 index 0000000..48f198e --- /dev/null +++ b/src/pyironscales/interfaces.py @@ -0,0 +1,102 @@ +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Generic, TypeVar + +from pyironscales.responses.paginated_response import PaginatedResponse +from pyironscales.types import ( + JSON, + IronscalesRequestParams, + PatchRequestData, +) + +if TYPE_CHECKING: + from pydantic import BaseModel + +TModel = TypeVar("TModel", bound="BaseModel") +TRequestParams = TypeVar( + "TRequestParams", + bound=IronscalesRequestParams, +) + + +class IMethodBase(ABC, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + self.model = model + + +class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def paginated( + self, + page: int, + page_size: int, + params: TRequestParams | None = None, + ) -> PaginatedResponse[TModel]: + pass + + +class IGettable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def get( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPostable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def post( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPatchable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def patch( + self, + data: PatchRequestData, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPuttable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def put( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IDeleteable(IMethodBase, Generic[TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def delete( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> None: + pass diff --git a/src/pyironscales/models/__init__.py b/src/pyironscales/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/models/base/__init__.py b/src/pyironscales/models/base/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/models/base/base_model.py b/src/pyironscales/models/base/base_model.py new file mode 100644 index 0000000..b08f427 --- /dev/null +++ b/src/pyironscales/models/base/base_model.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import inspect +from types import UnionType +from typing import Union, get_args, get_origin + +from pydantic import BaseModel, ConfigDict + +from pyironscales.utils.naming import to_camel_case + + +class IronscalesModel(BaseModel): + model_config = ConfigDict( + alias_generator=to_camel_case, + populate_by_name=True, + use_enum_values=True, + protected_namespaces=(), + ) + + @classmethod + def _get_field_names(cls) -> list[str]: + field_names = [] + for v in cls.__fields__.values(): + was_model = False + for arg in get_args(v.annotation): + if inspect.isclass(arg) and issubclass(arg, IronscalesModel): + was_model = True + field_names.extend([f"{v.alias}/{sub}" for sub in arg._get_field_names()]) + + if not was_model: + field_names.append(v.alias) + + return field_names + + @classmethod + def _get_field_names_and_types(cls) -> dict[str, str]: # noqa: C901 + field_names_and_types = {} + for v in cls.__fields__.values(): + was_model = False + field_type = "None" + if get_origin(v.annotation) is UnionType or get_origin(v.annotation) is Union: + for arg in get_args(v.annotation): + if inspect.isclass(arg) and issubclass(arg, IronscalesModel): + was_model = True + for sk, sv in arg._get_field_names_and_types().items(): + field_names_and_types[f"{v.alias}/{sk}"] = sv + elif arg is not None and arg.__name__ != "NoneType": + field_type = arg.__name__ + else: + if inspect.isclass(v.annotation) and issubclass(v.annotation, IronscalesModel): + was_model = True + for sk, sv in v.annotation._get_field_names_and_types().items(): + field_names_and_types[f"{v.alias}/{sk}"] = sv + elif v.annotation is not None and v.annotation.__name__ != "NoneType": + field_type = v.annotation.__name__ + + if not was_model: + field_names_and_types[v.alias] = field_type + + return field_names_and_types diff --git a/src/pyironscales/models/base/message_model.py b/src/pyironscales/models/base/message_model.py new file mode 100644 index 0000000..5bd7a02 --- /dev/null +++ b/src/pyironscales/models/base/message_model.py @@ -0,0 +1,5 @@ +from pydantic import BaseModel + + +class GenericMessageModel(BaseModel): + message: str diff --git a/src/pyironscales/models/ironscales/__init__.py b/src/pyironscales/models/ironscales/__init__.py new file mode 100644 index 0000000..661d26f --- /dev/null +++ b/src/pyironscales/models/ironscales/__init__.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import Field + +from pyironscales.models.base.base_model import IronscalesModel + +class Pagination(IronscalesModel): + current_page: int | None = Field(default=None, alias="CurrentPage") + current_page_count: int | None = Field(default=None, alias="CurrentPageCount") + limit: int | None = Field(default=None, alias="Limit") + total_count: int | None = Field(default=None, alias="TotalCount") + next_page: int | None = Field(default=None, alias="NextPage") + next_page_url: str | None = Field(default=None, alias="NextPageURL") + next_page_token: str | None = Field(default=None, alias="NextPageToken") + +class Answer(IronscalesModel): + id: int | None = Field(default=None, alias="Id") + created: datetime | None = Field(default=None, alias="Created") + modified: datetime | None = Field(default=None, alias="Modified") + question: dict[str, Any] | None = Field(default=None, alias="Question") + choice: str | None = Field(default=None, alias="Choice") + choice_label: str | None = Field(default=None, alias="ChoiceLabel") + choices: list | None = Field(default=None, alias="Choices") + sentiment: str | None = Field(default=None, alias="Sentiment") + comment: str | None = Field(default=None, alias="Comment") + follow_up_answer: str | None = Field(default=None, alias="FollowUpAnswer") + follow_up_answer_choice: str | None = Field(default=None, alias="FollowUpAnswerChoice") + follow_up_answer_choices: list | None = Field(default=None, alias="FollowUpAnswerChoices") + survey: dict[str, str | int] | None = Field(default=None, alias="Survey") + published_as_testimonial: bool | None = Field(default=None, alias="PublishedAsTestimonial") + response_id: int | None = Field(default=None, alias="ResponseId") + +class Customer(IronscalesModel): + id: int | None = Field(default=None, alias="Id") + external_id: str | None = Field(default=None, alias="ExternalId") + created: datetime | None = Field(default=None, alias="Created") + modified: datetime | None = Field(default=None, alias="Modified") + name: str | None = Field(default=None, alias="Name") + email: str | None = Field(default=None, alias="Email") + company: str | None = Field(default=None, alias="Company") + custom_attributes: dict[str, str | int] | None = Field(default=None, alias="CustomAttributes") + +class TeamMember(IronscalesModel): + id: int | None = Field(default=None, alias="Id") + external_id: str | None = Field(default=None, alias="ExternalId") + created: datetime | None = Field(default=None, alias="Created") + modified: datetime | None = Field(default=None, alias="Modified") + name: str | None = Field(default=None, alias="Name") + email: str | None = Field(default=None, alias="Email") + custom_attributes: dict[str, str | int] | None = Field(default=None, alias="CustomAttributes") + +class Response(IronscalesModel): + 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") + team_members: list[dict[str, Any]] | None = Field(default=None, alias="TeamMembers") + ticket: dict[str, Any] | None = Field(default=None, alias="Ticket") + customer: dict[str, Any] | None = Field(default=None, alias="Customer") + +class Survey(IronscalesModel): + id: int | None = Field(default=None, alias="Id") + name: str | None = Field(default=None, alias="Name") + metric: str | None = Field(default=None, alias="Metric") + survey_token: str | None = Field(default=None, alias="SurveyToken") + survey_type: str | None = Field(default=None, alias="SurveyType") + brand_name: str | None = Field(default=None, alias="BrandName") + +class CustomerBulk(IronscalesModel): + request_id: str | None = Field(default=None, alias="RequestId") + detail: str | None = Field(default=None, alias="Detail") + +class SurveyEmail(IronscalesModel): + detail: str | None = Field(default=None, alias="Detail") diff --git a/src/pyironscales/py.typed b/src/pyironscales/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/responses/__init__.py b/src/pyironscales/responses/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/responses/paginated_response.py b/src/pyironscales/responses/paginated_response.py new file mode 100644 index 0000000..483e79b --- /dev/null +++ b/src/pyironscales/responses/paginated_response.py @@ -0,0 +1,204 @@ +from __future__ import annotations +import json + +from typing import TYPE_CHECKING, Generic, TypeVar + +from pyironscales.utils.helpers import parse_link_headers, parse_response_body + +if TYPE_CHECKING: + from collections.abc import Iterable + + from pydantic import BaseModel + from requests import Response + + from pyironscales.types import RequestParams + + +TModel = TypeVar("TModel", bound="BaseModel") + +if TYPE_CHECKING: + from pyironscales.interfaces import IPaginateable + + +class PaginatedResponse(Generic[TModel]): + """ + PaginatedResponse is a wrapper class for handling paginated responses from the + Ironscales API. It provides methods for navigating through the pages of the response + and accessing the data contained within each page. + + The class is designed to work with IronscalesEndpoint and its derived classes to + parse the API response into model instances. It also supports iteration, allowing + the user to loop through the items within the paginated response. + + PaginatedResponse uses a generic type variable TModel, which represents the + expected model type for the response data. This allows for type-safe handling + of model instances throughout the class. + """ + + def __init__( + self, + response: Response, + response_model: type[TModel], + endpointmodel: IPaginateable, + endpoint: str, + page: int, + limit: int, + params: RequestParams | None = None, + ) -> None: + """ + PaginatedResponse is a wrapper class for handling paginated responses from the + Ironscales API. It provides methods for navigating through the pages of the response + and accessing the data contained within each page. + + The class is designed to work with IronscalesEndpoint and its derived classes to + parse the API response into model instances. It also supports iteration, allowing + the user to loop through the items within the paginated response. + + PaginatedResponse uses a generic type variable TModel, which represents the + 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) + + def _initialize( + self, + response: Response, + response_model: type[TModel], + endpointmodel: IPaginateable, + endpoint: str, + page: int, + limit: int, + params: RequestParams | None = None, + ): + """ + Initialize the instance variables using the provided response, endpointmodel, and page size. + + Args: + response: The raw response object from the API. + endpointmodel (IronscalesEndpoint[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')).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: + # Ironscales 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.has_data = self.data and len(self.data) > 0 + self.index = 0 + + def get_next_page(self) -> PaginatedResponse[TModel]: + """ + Fetch the next page of the paginated response. + + Returns: + PaginatedResponse[TModel]: The updated PaginatedResponse instance + with the data 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 + return self + + 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 + + def get_previous_page(self) -> PaginatedResponse[TModel]: + """ + Fetch the next page of the paginated response. + + Returns: + PaginatedResponse[TModel]: The updated PaginatedResponse instance + with the data 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 + return self + + 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 + + def all(self) -> Iterable[TModel]: + """ + Iterate through all items in the paginated response, across all pages. + + Yields: + TModel: An instance of the model class for each item in the paginated response. + """ + while self.has_data: + yield from self.data + self.get_next_page() + + def __iter__(self): + """ + Implement the iterator protocol for the PaginatedResponse class. + + Returns: + PaginatedResponse[TModel]: The current instance of the PaginatedResponse. + """ + return self + + def __dict__(self): + """ + Implement the iterator protocol for the PaginatedResponse class. + + Returns: + PaginatedResponse[TModel]: The current instance of the PaginatedResponse. + """ + return self.data + + def __next__(self): + """ + Implement the iterator protocol by getting the next item in the data. + + Returns: + TModel: The next item in the data. + + Raises: + StopIteration: If there are no more items in the data. + """ + if self.index < len(self.data): + result = self.data[self.index] + self.index += 1 + return result + else: + raise StopIteration diff --git a/src/pyironscales/types.py b/src/pyironscales/types.py new file mode 100644 index 0000000..09a428c --- /dev/null +++ b/src/pyironscales/types.py @@ -0,0 +1,42 @@ +from typing import Literal, TypeAlias + +from typing_extensions import NotRequired, TypedDict +from datetime import datetime + +Literals: TypeAlias = str | int | float | bool +JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | Literals | None + + +class Patch(TypedDict): + op: Literal["add"] | Literal["replace"] | Literal["remove"] + path: str + value: JSON + + +class IronscalesRequestParams(TypedDict): + created_at_min: NotRequired[datetime] + created_at_max: NotRequired[datetime] + updated_at_min: NotRequired[datetime] + updated_at_min: NotRequired[datetime] + customFieldConditions: NotRequired[str] + page_token: NotRequired[str] + page: NotRequired[int] + limit: NotRequired[int] + organization_id: NotRequired[int] + platform: NotRequired[str] + status: NotRequired[str] + indicator_type: NotRequired[str] + severity: NotRequired[str] + platform: NotRequired[str] + agent_id: NotRequired[str] + type: NotRequired[str] + entity_id: NotRequired[int] + types: NotRequired[str] + statuses: NotRequired[str] + + +GenericRequestParams: TypeAlias = dict[str, Literals] +RequestParams: TypeAlias = IronscalesRequestParams | GenericRequestParams +PatchRequestData: TypeAlias = list[Patch] +RequestData: TypeAlias = JSON | PatchRequestData +RequestMethod: TypeAlias = Literal["GET", "POST", "PUT", "PATCH", "DELETE"] diff --git a/src/pyironscales/utils/__init__.py b/src/pyironscales/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/utils/experimental/__init__.py b/src/pyironscales/utils/experimental/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pyironscales/utils/experimental/condition.py b/src/pyironscales/utils/experimental/condition.py new file mode 100644 index 0000000..98980cf --- /dev/null +++ b/src/pyironscales/utils/experimental/condition.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import inspect +import re +from datetime import datetime +from enum import Enum +from typing import TYPE_CHECKING, Any, Generic, TypeVar + +from pyironscales.utils.naming import to_camel_case + +if TYPE_CHECKING: + from collections.abc import Callable + +T = TypeVar("T") + + +class ValueType(Enum): + STR = 1 + INT = 2 + DATETIME = 3 + + +class Condition(Generic[T]): + def __init__(self: Condition[T]) -> None: + self._condition_string: str = "" + self._field = "" + + def field(self: Condition[T], selector: Callable[[type[T]], Any]) -> Condition[T]: + field = "" + + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"field\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = to_camel_case("/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:])) + + finally: + del frame + + self._condition_string += field + return self + + def equals(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " = " + self.__add_typed_value_to_string(value, type(value)) + return self + + def not_equals(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " = " + self.__add_typed_value_to_string(value, type(value)) + return self + + def less_than(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " < " + self.__add_typed_value_to_string(value, type(value)) + return self + + def less_than_or_equals( + self: Condition[T], + value: Any, # noqa: ANN401 + ) -> Condition[T]: + self._condition_string += " <= " + self.__add_typed_value_to_string(value, type(value)) + return self + + def greater_than(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " > " + self.__add_typed_value_to_string(value, type(value)) + return self + + def greater_than_or_equals( + self: Condition[T], + value: Any, # noqa: ANN401 + ) -> Condition[T]: + self._condition_string += " >= " + self.__add_typed_value_to_string(value, type(value)) + return self + + def contains(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " CONTAINS " + self.__add_typed_value_to_string(value, type(value)) + return self + + def like(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " LIKE " + self.__add_typed_value_to_string(value, type(value)) + return self + + def in_(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " IN " + self.__add_typed_value_to_string(value, type(value)) + return self + + def not_(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " NOT " + self.__add_typed_value_to_string(value, type(value)) + return self + + def __add_typed_value_to_string( # noqa: ANN202 + self: Condition[T], + value: Any, # noqa: ANN401 + type: type, # noqa: A002 + ): + if type is str: + self._condition_string += f'"{value}"' + elif type is int: # noqa: SIM114 + self._condition_string += str(value) + elif type is bool: + self._condition_string += str(value) + elif type is datetime: + self._condition_string += f"[{value}]" + else: + self._condition_string += f'"{value}"' + + def and_(self: Condition[T], selector: Callable[[type[T]], Any] | None = None) -> Condition[T]: + self._condition_string += " AND " + + if selector is not None: + field = "" + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"and_\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = "/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:]) + + finally: + del frame + + self._condition_string += field + return self + + def or_(self: Condition[T], selector: Callable[[type[T]], Any] | None = None) -> Condition[T]: + self._condition_string += " OR " + + if selector is not None: + field = "" + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"or_\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = "/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:]) + + finally: + del frame + + self._condition_string += field + return self + + def wrap(self: Condition[T], condition: Callable[[Condition[T]], Condition[T]]) -> Condition[T]: + self._condition_string += f"({condition(Condition[T]())})" + return self + + def __str__(self: Condition[T]) -> str: + return self._condition_string.strip() diff --git a/src/pyironscales/utils/experimental/patch_maker.py b/src/pyironscales/utils/experimental/patch_maker.py new file mode 100644 index 0000000..4c673ce --- /dev/null +++ b/src/pyironscales/utils/experimental/patch_maker.py @@ -0,0 +1,37 @@ +import json +from enum import Enum +from typing import Any + + +class Patch: + class PatchOp(Enum): + """ + PatchOperation is an enumeration of the different patch operations supported + by the Ironscales API. These operations are ADD, REPLACE, and REMOVE. + """ + + ADD = 1 + REPLACE = 2 + REMOVE = 3 + + def __init__(self, op: PatchOp, path: str, value: Any) -> None: # noqa: ANN401 + self.op = op.name.lower() + self.path = path + self.value = value + + def __repr__(self) -> str: + """ + Return a string representation of the model as a formatted JSON string. + + Returns: + str: A formatted JSON string representation of the model. + """ + return json.dumps(self.__dict__, default=str, indent=2) + + +class PatchGroup: + def __init__(self, *patches: Patch) -> None: + self.patches = list(patches) + + def __repr__(self) -> str: + return str(self.patches) diff --git a/src/pyironscales/utils/helpers.py b/src/pyironscales/utils/helpers.py new file mode 100644 index 0000000..56239f8 --- /dev/null +++ b/src/pyironscales/utils/helpers.py @@ -0,0 +1,190 @@ +import re +import math +from datetime import datetime +from typing import Any + +from requests.structures import CaseInsensitiveDict + + +def cw_format_datetime(dt: datetime) -> str: + """Format a datetime object as a string in ISO 8601 format. This is the format that Ironscales uses. + + Args: + dt (datetime): The datetime object to be formatted. + + Returns: + str: The formatted datetime string in the format "YYYY-MM-DDTHH:MM:SSZ". + + Example: + from datetime import datetime + + dt = datetime(2022, 1, 1, 12, 0, 0) + formatted_dt = cw_format_datetime(dt) + print(formatted_dt) # Output: "2022-01-01T12:00:00Z" + """ + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + +def parse_response_body( + body: CaseInsensitiveDict, +) -> dict[str, Any] | None: + """ + Parses response body to extract pagination information. + + Arguments: + - body: content.json().get('pagination', {}) A dictionary containing the headers of an HTTP response. + + Returns: + - A dictionary containing the extracted pagination information. The keys in the dictionary include: + - "first_page": An optional integer representing the number of the first page. + - "prev_page": An optional integer representing the number of the previous page. + - "next_page": An optional integer representing the number of the next page. + - "last_page": An optional integer representing the number of the last page. + - "has_next_page": A boolean indicating whether there is a next page. + - "has_prev_page": A boolean indicating whether there is a previous page. + + If the "Link" header is not present in the headers dictionary, None is returned. + + Example Usage: + headers = { + "Link": '; rel="first", ; rel="next"' + } + pagination_info = parse_link_headers(headers) + print(pagination_info) + # Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True} + """ + if body.get("current_page") is None: + return None + has_next_page: bool = False + has_prev_page: bool = False + first_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_url: str | None = None + next_page_token: str | None = None + last_page: int | None = None + + result = {} + + if body.get("first_page") is not None: + result["first_page"] = body.get("first_page") + + if body.get("prev_page") 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 + elif body.get("current_page") is not None and body.get("next_page") is None: + 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"): + 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 + + return result + +def parse_link_headers( + headers: CaseInsensitiveDict, +) -> dict[str, Any] | None: + """ + Parses link headers to extract pagination information. + + Arguments: + - headers: A dictionary containing the headers of an HTTP response. The value associated with the "Link" key should be a string representing the link headers. + + Returns: + - A dictionary containing the extracted pagination information. The keys in the dictionary include: + - "first_page": An optional integer representing the number of the first page. + - "prev_page": An optional integer representing the number of the previous page. + - "next_page": An optional integer representing the number of the next page. + - "last_page": An optional integer representing the number of the last page. + - "has_next_page": A boolean indicating whether there is a next page. + - "has_prev_page": A boolean indicating whether there is a previous page. + + If the "Link" header is not present in the headers dictionary, None is returned. + + Example Usage: + headers = { + "Link": '; rel="first", ; rel="next"' + } + pagination_info = parse_link_headers(headers) + print(pagination_info) + # Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True} + """ + if headers.get("Link") is None: + return None + links = headers["Link"].split(",") + has_next_page: bool = False + has_prev_page: bool = False + first_page: int | None = None + prev_page: int | None = None + next_page: int | None = None + last_page: int | None = None + + for link in links: + match = re.search(r'page=(\d+)>; rel="(.*?)"', link) + if match: + page_number = int(match.group(1)) + rel_value = match.group(2) + if rel_value == "first": + first_page = page_number + elif rel_value == "prev": + prev_page = page_number + has_prev_page = True + elif rel_value == "next": + next_page = page_number + has_next_page = True + elif rel_value == "last": + last_page = page_number + + result = {} + + if first_page is not None: + result["first_page"] = first_page + + if prev_page is not None: + result["prev_page"] = prev_page + + if next_page is not None: + result["next_page"] = next_page + + if last_page is not None: + result["last_page"] = last_page + + if has_next_page: + result["has_next_page"] = has_next_page + + if has_prev_page: + result["has_prev_page"] = has_prev_page + + return result diff --git a/src/pyironscales/utils/naming.py b/src/pyironscales/utils/naming.py new file mode 100644 index 0000000..e0d1758 --- /dev/null +++ b/src/pyironscales/utils/naming.py @@ -0,0 +1,23 @@ +from keyword import iskeyword + + +def to_snake_case(string: str) -> str: + return ("_" if string.startswith("_") else "") + "".join( + ["_" + i.lower() if i.isupper() else i for i in string.lstrip("_")] + ).lstrip("_") + + +def to_camel_case(string: str) -> str: + string_split = string.split("_") + return string_split[0] + "".join(word.capitalize() for word in string_split[1:]) + + +def to_title_case_preserve_case(string: str) -> str: + return string[:1].upper() + string[1:] + + +def ensure_not_reserved(string: str) -> str: + if iskeyword(string): + return string + "_" + else: # noqa: RET505 + return string