mirror of
https://github.com/brygphilomena/pyironscales.git
synced 2025-12-05 23:12:34 +00:00
Lets get started
This commit is contained in:
parent
786793bb4c
commit
83a238d206
57 changed files with 2629 additions and 0 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -205,3 +205,4 @@ cython_debug/
|
|||
marimo/_static/
|
||||
marimo/_lsp/
|
||||
__marimo__/
|
||||
src/ironscales_scratchpad.py
|
||||
|
|
|
|||
39
pyproject.toml
Normal file
39
pyproject.toml
Normal file
|
|
@ -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"]
|
||||
3
requirements.txt
Normal file
3
requirements.txt
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
requests==2.32.4
|
||||
pydantic==2.11.7
|
||||
typing_extensions==4.14.1
|
||||
4
src/pyironscales/__init__.py
Normal file
4
src/pyironscales/__init__.py
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
from pyironscales.clients.ironscales_client import IronscalesAPIClient
|
||||
|
||||
__all__ = ["IronscalesAPIClient"]
|
||||
__version__ = "0.1.1"
|
||||
0
src/pyironscales/clients/__init__.py
Normal file
0
src/pyironscales/clients/__init__.py
Normal file
133
src/pyironscales/clients/base_client.py
Normal file
133
src/pyironscales/clients/base_client.py
Normal file
|
|
@ -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
|
||||
93
src/pyironscales/clients/ironscales_client.py
Normal file
93
src/pyironscales/clients/ironscales_client.py
Normal file
|
|
@ -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}",
|
||||
}
|
||||
9
src/pyironscales/config.py
Normal file
9
src/pyironscales/config.py
Normal file
|
|
@ -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
|
||||
0
src/pyironscales/endpoints/__init__.py
Normal file
0
src/pyironscales/endpoints/__init__.py
Normal file
0
src/pyironscales/endpoints/base/__init__.py
Normal file
0
src/pyironscales/endpoints/base/__init__.py
Normal file
163
src/pyironscales/endpoints/base/base_endpoint.py
Normal file
163
src/pyironscales/endpoints/base/base_endpoint.py
Normal file
|
|
@ -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)
|
||||
24
src/pyironscales/endpoints/ironscales/AnswersEndpoint.py
Normal file
24
src/pyironscales/endpoints/ironscales/AnswersEndpoint.py
Normal file
|
|
@ -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
|
||||
59
src/pyironscales/endpoints/ironscales/AnswersIdEndpoint.py
Normal file
59
src/pyironscales/endpoints/ironscales/AnswersIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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', {}))
|
||||
|
|
@ -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())
|
||||
46
src/pyironscales/endpoints/ironscales/CustomersEndpoint.py
Normal file
46
src/pyironscales/endpoints/ironscales/CustomersEndpoint.py
Normal file
|
|
@ -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())
|
||||
59
src/pyironscales/endpoints/ironscales/CustomersIdEndpoint.py
Normal file
59
src/pyironscales/endpoints/ironscales/CustomersIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
38
src/pyironscales/endpoints/ironscales/QuestionsEndpoint.py
Normal file
38
src/pyironscales/endpoints/ironscales/QuestionsEndpoint.py
Normal file
|
|
@ -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', {}),
|
||||
)
|
||||
|
|
@ -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())
|
||||
26
src/pyironscales/endpoints/ironscales/ResponsesEndpoint.py
Normal file
26
src/pyironscales/endpoints/ironscales/ResponsesEndpoint.py
Normal file
|
|
@ -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
|
||||
37
src/pyironscales/endpoints/ironscales/ResponsesIdEndpoint.py
Normal file
37
src/pyironscales/endpoints/ironscales/ResponsesIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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', {}))
|
||||
52
src/pyironscales/endpoints/ironscales/SurveysEndpoint.py
Normal file
52
src/pyironscales/endpoints/ironscales/SurveysEndpoint.py
Normal file
|
|
@ -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', {}),
|
||||
)
|
||||
|
|
@ -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())
|
||||
10
src/pyironscales/endpoints/ironscales/SurveysIdEndpoint.py
Normal file
10
src/pyironscales/endpoints/ironscales/SurveysIdEndpoint.py
Normal file
|
|
@ -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))
|
||||
44
src/pyironscales/endpoints/ironscales/TeamMembersEndpoint.py
Normal file
44
src/pyironscales/endpoints/ironscales/TeamMembersEndpoint.py
Normal file
|
|
@ -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())
|
||||
|
|
@ -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(),
|
||||
)
|
||||
0
src/pyironscales/endpoints/ironscales/__init__.py
Normal file
0
src/pyironscales/endpoints/ironscales/__init__.py
Normal file
59
src/pyironscales/endpoints/simplesat/AnswersIdEndpoint.py
Normal file
59
src/pyironscales/endpoints/simplesat/AnswersIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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', {}))
|
||||
|
|
@ -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())
|
||||
59
src/pyironscales/endpoints/simplesat/CustomersIdEndpoint.py
Normal file
59
src/pyironscales/endpoints/simplesat/CustomersIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
38
src/pyironscales/endpoints/simplesat/QuestionsEndpoint.py
Normal file
38
src/pyironscales/endpoints/simplesat/QuestionsEndpoint.py
Normal file
|
|
@ -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', {}),
|
||||
)
|
||||
|
|
@ -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())
|
||||
37
src/pyironscales/endpoints/simplesat/ResponsesIdEndpoint.py
Normal file
37
src/pyironscales/endpoints/simplesat/ResponsesIdEndpoint.py
Normal file
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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(),
|
||||
)
|
||||
|
|
@ -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', {}))
|
||||
|
|
@ -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())
|
||||
|
|
@ -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(),
|
||||
)
|
||||
89
src/pyironscales/exceptions.py
Normal file
89
src/pyironscales/exceptions.py
Normal file
|
|
@ -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."
|
||||
102
src/pyironscales/interfaces.py
Normal file
102
src/pyironscales/interfaces.py
Normal file
|
|
@ -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
|
||||
0
src/pyironscales/models/__init__.py
Normal file
0
src/pyironscales/models/__init__.py
Normal file
0
src/pyironscales/models/base/__init__.py
Normal file
0
src/pyironscales/models/base/__init__.py
Normal file
60
src/pyironscales/models/base/base_model.py
Normal file
60
src/pyironscales/models/base/base_model.py
Normal file
|
|
@ -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
|
||||
5
src/pyironscales/models/base/message_model.py
Normal file
5
src/pyironscales/models/base/message_model.py
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class GenericMessageModel(BaseModel):
|
||||
message: str
|
||||
76
src/pyironscales/models/ironscales/__init__.py
Normal file
76
src/pyironscales/models/ironscales/__init__.py
Normal file
|
|
@ -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")
|
||||
0
src/pyironscales/py.typed
Normal file
0
src/pyironscales/py.typed
Normal file
0
src/pyironscales/responses/__init__.py
Normal file
0
src/pyironscales/responses/__init__.py
Normal file
204
src/pyironscales/responses/paginated_response.py
Normal file
204
src/pyironscales/responses/paginated_response.py
Normal file
|
|
@ -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
|
||||
42
src/pyironscales/types.py
Normal file
42
src/pyironscales/types.py
Normal file
|
|
@ -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"]
|
||||
0
src/pyironscales/utils/__init__.py
Normal file
0
src/pyironscales/utils/__init__.py
Normal file
0
src/pyironscales/utils/experimental/__init__.py
Normal file
0
src/pyironscales/utils/experimental/__init__.py
Normal file
166
src/pyironscales/utils/experimental/condition.py
Normal file
166
src/pyironscales/utils/experimental/condition.py
Normal file
|
|
@ -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()
|
||||
37
src/pyironscales/utils/experimental/patch_maker.py
Normal file
37
src/pyironscales/utils/experimental/patch_maker.py
Normal file
|
|
@ -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)
|
||||
190
src/pyironscales/utils/helpers.py
Normal file
190
src/pyironscales/utils/helpers.py
Normal file
|
|
@ -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": '<https://example.com/api?page=1>; rel="first", <https://example.com/api?page=2>; 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": '<https://example.com/api?page=1>; rel="first", <https://example.com/api?page=2>; 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
|
||||
23
src/pyironscales/utils/naming.py
Normal file
23
src/pyironscales/utils/naming.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue