diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2fdab44 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,39 @@ +[project] +name = "pysimplesat" +version = "0.1.1" +authors = [ + { name="Peter Annabel", email="peter.annabel@gmail.com" }, +] +description = "A full-featured Python client for the SimpleSat 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 = [ + "SimpleSat", + "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/pysimplesat" +Issues = "https://github.com/brygphilomena/pysimplesat/issues" + +[build-system] +requires = ["hatchling >= 1.26", "hatch-requirements-txt"] +build-backend = "hatchling.build" + +[tool.hatch.metadata.hooks.requirements_txt] +files = ["requirements.txt"] \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..04189d9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests==2.32.4 +pydantic==2.11.7 +typing_extensions==4.14.1 diff --git a/src/pysimplesat/__init__.py b/src/pysimplesat/__init__.py new file mode 100644 index 0000000..4ef84bb --- /dev/null +++ b/src/pysimplesat/__init__.py @@ -0,0 +1,4 @@ +from pysimplesat.clients.simplesat_client import SimpleSatAPIClient + +__all__ = ["SimpleSatAPIClient"] +__version__ = "0.1.1" diff --git a/src/pysimplesat/clients/__init__.py b/src/pysimplesat/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/clients/base_client.py b/src/pysimplesat/clients/base_client.py new file mode 100644 index 0000000..8328ffc --- /dev/null +++ b/src/pysimplesat/clients/base_client.py @@ -0,0 +1,134 @@ +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 pysimplesat.config import Config +from pysimplesat.exceptions import ( + AuthenticationFailedException, + ConflictException, + MalformedRequestException, + MethodNotAllowedException, + NotFoundException, + ObjectExistsError, + PermissionsFailedException, + ServerError, + TooManyRequestsException, +) + +if TYPE_CHECKING: + from pysimplesat.types import RequestData, RequestMethod, RequestParams + + +class SimpleSatClient(ABC): + config: Config = Config() + + @abstractmethod + def _get_headers(self) -> dict[str, str]: + pass + + @abstractmethod + def _get_url(self) -> str: + pass + + def _make_request( # noqa: C901 + self, + method: RequestMethod, + url: str, + data: RequestData | None = None, +# rawdata: RequestData | None = None, + params: RequestParams | None = None, + headers: dict[str, str] | None = None, + retry_count: int = 0, + 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. + """ + + 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, + data=data, + params=cast(dict[str, Any], params or {}), + stream=stream, + ) + else: + response = requests.request( + method, + url, + headers=headers, + params=cast(dict[str, Any], params or {}), + stream=stream, + ) + if not response.ok: + with contextlib.suppress(json.JSONDecodeError): + details: dict = response.json() + if response.status_code == 400: + if details.get("code") == "InvalidObject": + errors = details.get("errors", []) + if len(errors) > 1: + warnings.warn( + "Found multiple errors - we may be masking some important error details. Please submit a Github issue with response.status_code and response.content so we can improve this error handling.", + stacklevel=1, + ) + for error in errors: + if error.get("code") == "ObjectExists": + error.pop("code") # Don't need code in message + raise ObjectExistsError(response, extra_message=json.dumps(error, indent=4)) + + if response.status_code == 400: + raise MalformedRequestException(response) + if response.status_code == 401: + raise AuthenticationFailedException(response) + if response.status_code == 403: + raise PermissionsFailedException(response) + if response.status_code == 404: + raise NotFoundException(response) + if response.status_code == 405: + raise MethodNotAllowedException(response) + if response.status_code == 409: + raise ConflictException(response) + if response.status_code == 429: + raise TooManyRequestsException(response) + if response.status_code == 500: + # if timeout is mentioned anywhere in the response then we'll retry. + # Ideally we'd return immediately on any non-timeout errors (since + # retries won't help much there), but err towards classifying too much + # as retries instead of too little. + if "timeout" in (response.text + response.reason).lower(): + if retry_count < self.config.max_retries: + retry_count += 1 + return self._make_request(method, url, data, params, headers, retry_count) + raise Timeout(response=response) + raise ServerError(response) + + return response diff --git a/src/pysimplesat/clients/simplesat_client.py b/src/pysimplesat/clients/simplesat_client.py new file mode 100644 index 0000000..b34c16b --- /dev/null +++ b/src/pysimplesat/clients/simplesat_client.py @@ -0,0 +1,106 @@ +import typing +from datetime import datetime, timezone +import base64 + +from pysimplesat.clients.base_client import SimpleSatClient +from pysimplesat.config import Config + +if typing.TYPE_CHECKING: + from pysimplesat.endpoints.simplesat.AccountEndpoint import AccountEndpoint + from pysimplesat.endpoints.simplesat.ActorEndpoint import ActorEndpoint + from pysimplesat.endpoints.simplesat.AgentsEndpoint import AgentsEndpoint + from pysimplesat.endpoints.simplesat.BillingreportsEndpoint import BillingreportsEndpoint + from pysimplesat.endpoints.simplesat.IncidentreportsEndpoint import IncidentreportsEndpoint + from pysimplesat.endpoints.simplesat.OrganizationsEndpoint import OrganizationsEndpoint + from pysimplesat.endpoints.simplesat.ReportsEndpoint import ReportsEndpoint + from pysimplesat.endpoints.simplesat.SignalsEndpoint import SignalsEndpoint + + +class SimpleSatAPIClient(SimpleSatClient): + """ + SimpleSat API client. Handles the connection to the SimpleSat API + and the configuration of all the available endpoints. + """ + + def __init__( + self, + privatekey: str, + ) -> None: + """ + Initializes the client with the given credentials. + + Parameters: + privatekey (str): Your SimpleSat API private key. + """ + self.privatekey: str = privatekey + self.token_expiry_time: datetime = datetime.now(tz=timezone.utc) + + # Initializing endpoints + @property + def account(self) -> "AccountEndpoint": + from pysimplesat.endpoints.simplesat.AccountEndpoint import AccountEndpoint + + return AccountEndpoint(self) + + @property + def actor(self) -> "ActorEndpoint": + from pysimplesat.endpoints.simplesat.ActorEndpoint import ActorEndpoint + + return ActorEndpoint(self) + + @property + def agents(self) -> "AgentsEndpoint": + from pysimplesat.endpoints.simplesat.AgentsEndpoint import AgentsEndpoint + + return AgentsEndpoint(self) + + @property + def billing_reports(self) -> "BillingreportsEndpoint": + from pysimplesat.endpoints.simplesat.BillingreportsEndpoint import BillingreportsEndpoint + + return BillingreportsEndpoint(self) + + @property + def incident_reports(self) -> "IncidentreportsEndpoint": + from pysimplesat.endpoints.simplesat.IncidentreportsEndpoint import IncidentreportsEndpoint + + return IncidentreportsEndpoint(self) + + @property + def organizations(self) -> "OrganizationsEndpoint": + from pysimplesat.endpoints.simplesat.OrganizationsEndpoint import OrganizationsEndpoint + + return OrganizationsEndpoint(self) + + @property + def reports(self) -> "ReportsEndpoint": + from pysimplesat.endpoints.simplesat.ReportsEndpoint import ReportsEndpoint + + return ReportsEndpoint(self) + + @property + def signals(self) -> "SignalsEndpoint": + from pysimplesat.endpoints.simplesat.SignalsEndpoint import SignalsEndpoint + + return SignalsEndpoint(self) + + def _get_url(self) -> str: + """ + Generates and returns the URL for the SimpleSat API endpoints based on the company url and codebase. + Logs in an obtains an access token. + Returns: + str: API URL. + """ + return f"https://api.simplesat.io/api/v1" + + 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", + "X-Simplesat-Token": f"{self.privatekey}", + } diff --git a/src/pysimplesat/config.py b/src/pysimplesat/config.py new file mode 100644 index 0000000..e9c2435 --- /dev/null +++ b/src/pysimplesat/config.py @@ -0,0 +1,9 @@ +class Config: + def __init__(self, max_retries=3) -> None: + """ + Initializes a new instance of the Config class. + + Args: + max_retries (int): The maximum number of retries for a retryable HTTP operation (500) (default = 3) + """ + self.max_retries = max_retries diff --git a/src/pysimplesat/endpoints/__init__.py b/src/pysimplesat/endpoints/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/endpoints/base/__init__.py b/src/pysimplesat/endpoints/base/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/endpoints/base/base_endpoint.py b/src/pysimplesat/endpoints/base/base_endpoint.py new file mode 100644 index 0000000..47c4382 --- /dev/null +++ b/src/pysimplesat/endpoints/base/base_endpoint.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, TypeVar + +if TYPE_CHECKING: + from pydantic import BaseModel + from requests import Response + + from pysimplesat.clients.base_client import SimpleSatClient + from pysimplesat.types import ( + RequestData, + RequestMethod, + RequestParams, + ) + +TChildEndpoint = TypeVar("TChildEndpoint", bound="SimpleSatEndpoint") +TModel = TypeVar("TModel", bound="BaseModel") + + +class SimpleSatEndpoint: + """ + SimpleSatEndpoint is a base class for all SimpleSat API endpoint classes. + It provides a generic implementation for interacting with the SimpleSat API, + handling requests, parsing responses into model instances, and managing pagination. + + SimpleSatEndpoint makes use of a generic type variable TModel, which represents + the expected SimpleSatModel type for the endpoint. This allows for type-safe + handling of model instances throughout the class. + + Each derived class should specify the SimpleSatModel type it will be working with + when inheriting from SimpleSatEndpoint. For example: + class CompanyEndpoint(SimpleSatEndpoint[CompanyModel]). + + SimpleSatEndpoint 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. + + SimpleSatEndpoint 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 SimpleSatAPIClient instance. + endpoint_url (str): The base URL for the specific endpoint. + parent_endpoint (SimpleSatEndpoint, optional): The parent endpoint, if applicable. + + Attributes: + client (SimpleSatAPIClient): The SimpleSatAPIClient instance. + endpoint_url (str): The base URL for the specific endpoint. + _parent_endpoint (SimpleSatEndpoint): 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[SimpleSatEndpoint]): A list of registered child endpoints. + + Generic Type: + TModel: The model class for the endpoint. + """ + + def __init__( + self, + client: SimpleSatClient, + endpoint_url: str, + parent_endpoint: SimpleSatEndpoint | None = None, + ) -> None: + """ + Initialize a SimpleSatEndpoint instance with the client and endpoint base. + + Args: + client: The SimpleSatAPIClient 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[SimpleSatEndpoint] = [] + + def _register_child_endpoint(self, child_endpoint: TChildEndpoint) -> TChildEndpoint: + """ + Register a child endpoint to the current endpoint. + + Args: + child_endpoint (SimpleSatEndpoint): The child endpoint instance. + + Returns: + SimpleSatEndpoint: 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: SimpleSatEndpoint | 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: SimpleSatEndpoint) -> str: + if other_endpoint._parent_endpoint is not None: + parent_url = self._build_url(other_endpoint._parent_endpoint) + if other_endpoint._parent_endpoint._id is not None: + return self._url_join( + parent_url, + other_endpoint._get_replaced_url(), + ) + else: # noqa: RET505 + return self._url_join(parent_url, other_endpoint._get_replaced_url()) + else: + return self._url_join(self.client._get_url(), other_endpoint._get_replaced_url()) + + def _get_endpoint_url(self) -> str: + return self._build_url(self) + + def _parse_many(self, model_type: type[TModel], data: list[dict[str, Any]]) -> list[TModel]: + return [model_type.model_validate(d) for d in data] + + def _parse_one(self, model_type: type[TModel], data: dict[str, Any]) -> TModel: + return model_type.model_validate(data) diff --git a/src/pysimplesat/endpoints/siem/AccountEndpoint.py b/src/pysimplesat/endpoints/siem/AccountEndpoint.py new file mode 100644 index 0000000..542e2f9 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/AccountEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import Account +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class AccountEndpoint( + SimpleSatEndpoint, + IGettable[Account, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "account", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Account) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Account: + """ + Performs a GET request against the /account 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( + Account, + super()._make_request("GET", data=data, params=params).json().get('account', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/ActorEndpoint.py b/src/pysimplesat/endpoints/siem/ActorEndpoint.py new file mode 100644 index 0000000..7c841c7 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/ActorEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import ActorResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class ActorEndpoint( + SimpleSatEndpoint, + IGettable[ActorResponse, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "actor", parent_endpoint=parent_endpoint) + IGettable.__init__(self, ActorResponse) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> ActorResponse: + """ + Performs a GET request against the /Actor 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( + ActorResponse, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pysimplesat/endpoints/siem/AgentsIdEndpoint.py b/src/pysimplesat/endpoints/siem/AgentsIdEndpoint.py new file mode 100644 index 0000000..3b4bb54 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/AgentsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Agents +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class AgentsIdEndpoint( + SimpleSatEndpoint, + IGettable[Agents, SimpleSatRequestParams], + IPaginateable[Agents, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Agents) + IPaginateable.__init__(self, Agents) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Agents]: + """ + Performs a GET request against the /agents 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[Agents]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Agents, + self, + "agents", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Agents: + """ + Performs a GET request against the /agents 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( + Agents, + super()._make_request("GET", data=data, params=params).json().get('agent', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/BillingreportsIdEndpoint.py b/src/pysimplesat/endpoints/siem/BillingreportsIdEndpoint.py new file mode 100644 index 0000000..9fc9775 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/BillingreportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import BillingReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class BillingIdreportsEndpoint( + SimpleSatEndpoint, + IGettable[BillingReports, SimpleSatRequestParams], + IPaginateable[BillingReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, BillingReports) + IPaginateable.__init__(self, BillingReports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[BillingReports]: + """ + Performs a GET request against the /billing_reports 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[BillingReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + BillingReports, + self, + "billing_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> BillingReports: + """ + Performs a GET request against the /billing_reports 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( + BillingReports, + super()._make_request("GET", data=data, params=params).json().get('billing_report', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/IncidentreportsIdEndpoint.py b/src/pysimplesat/endpoints/siem/IncidentreportsIdEndpoint.py new file mode 100644 index 0000000..5a3c3f0 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/IncidentreportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import IncidentReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class IncidentreportsIdEndpoint( + SimpleSatEndpoint, + IGettable[IncidentReports, SimpleSatRequestParams], + IPaginateable[IncidentReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, IncidentReports) + IPaginateable.__init__(self, IncidentReports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[IncidentReports]: + """ + Performs a GET request against the /incident_reports 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[IncidentReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + IncidentReports, + self, + "incident_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> IncidentReports: + """ + Performs a GET request against the /incident_reports 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( + IncidentReports, + super()._make_request("GET", data=data, params=params).json().get('incident_report', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/OrganizationsIdEndpoint.py b/src/pysimplesat/endpoints/siem/OrganizationsIdEndpoint.py new file mode 100644 index 0000000..19b480a --- /dev/null +++ b/src/pysimplesat/endpoints/siem/OrganizationsIdEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import Organizations +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class OrganizationsIdEndpoint( + SimpleSatEndpoint, + IGettable[Organizations, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Organizations) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Organizations: + """ + Performs a GET request against the /organizations/{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( + Organizations, + super()._make_request("GET", data=data, params=params).json().get('organization', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/ReportsIdEndpoint.py b/src/pysimplesat/endpoints/siem/ReportsIdEndpoint.py new file mode 100644 index 0000000..97f59da --- /dev/null +++ b/src/pysimplesat/endpoints/siem/ReportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Reports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class ReportsIdEndpoint( + SimpleSatEndpoint, + IGettable[Reports, SimpleSatRequestParams], + IPaginateable[Reports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Reports) + IPaginateable.__init__(self, Reports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Reports]: + """ + Performs a GET request against the /reports 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[Reports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Reports, + self, + "reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Reports: + """ + Performs a GET request against the /reports 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( + Reports, + super()._make_request("GET", data=data, params=params).json().get('report', {}), + ) diff --git a/src/pysimplesat/endpoints/siem/SignalsIdEndpoint.py b/src/pysimplesat/endpoints/siem/SignalsIdEndpoint.py new file mode 100644 index 0000000..e3075e7 --- /dev/null +++ b/src/pysimplesat/endpoints/siem/SignalsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Signals +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class SignalsIdEndpoint( + SimpleSatEndpoint, + IGettable[Signals, SimpleSatRequestParams], + IPaginateable[Signals, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Signals) + IPaginateable.__init__(self, Signals) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Signals]: + """ + Performs a GET request against the /signals 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[Signals]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Signals, + self, + "signals", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Signals: + """ + Performs a GET request against the /signals 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( + Signals, + super()._make_request("GET", data=data, params=params).json().get('signal', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/AccountEndpoint.py b/src/pysimplesat/endpoints/simplesat/AccountEndpoint.py new file mode 100644 index 0000000..034e9fe --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/AccountEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import BaseEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import Account +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class AccountEndpoint( + SimpleSatEndpoint, + IGettable[Account, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "account", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Account) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Account: + """ + Performs a GET request against the /account 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( + Account, + super()._make_request("GET", data=data, params=params).json().get('account', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/ActorEndpoint.py b/src/pysimplesat/endpoints/simplesat/ActorEndpoint.py new file mode 100644 index 0000000..7c841c7 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/ActorEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import ActorResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class ActorEndpoint( + SimpleSatEndpoint, + IGettable[ActorResponse, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "actor", parent_endpoint=parent_endpoint) + IGettable.__init__(self, ActorResponse) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> ActorResponse: + """ + Performs a GET request against the /Actor 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( + ActorResponse, + super()._make_request("GET", data=data, params=params).json(), + ) diff --git a/src/pysimplesat/endpoints/simplesat/AgentsEndpoint.py b/src/pysimplesat/endpoints/simplesat/AgentsEndpoint.py new file mode 100644 index 0000000..2ed7914 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/AgentsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.AgentsIdEndpoint import AgentsIdEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Agents +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class AgentsEndpoint( + SimpleSatEndpoint, + IGettable[Agents, SimpleSatRequestParams], + IPaginateable[Agents, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "agents", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Agents) + IPaginateable.__init__(self, Agents) + + def id(self, id: int) -> AgentsIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized AgentsIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + AgentsIdEndpoint: The initialized AgentsIdEndpoint object. + """ + child = AgentsIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Agents]: + """ + Performs a GET request against the /agents 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[Agents]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Agents, + self, + "agents", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Agents: + """ + Performs a GET request against the /agents 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_many( + Agents, + super()._make_request("GET", data=data, params=params).json().get('agents', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/AgentsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/AgentsIdEndpoint.py new file mode 100644 index 0000000..3b4bb54 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/AgentsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Agents +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class AgentsIdEndpoint( + SimpleSatEndpoint, + IGettable[Agents, SimpleSatRequestParams], + IPaginateable[Agents, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Agents) + IPaginateable.__init__(self, Agents) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Agents]: + """ + Performs a GET request against the /agents 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[Agents]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Agents, + self, + "agents", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Agents: + """ + Performs a GET request against the /agents 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( + Agents, + super()._make_request("GET", data=data, params=params).json().get('agent', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/BillingreportsEndpoint.py b/src/pysimplesat/endpoints/simplesat/BillingreportsEndpoint.py new file mode 100644 index 0000000..9ed9509 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/BillingreportsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.BillingreportsIdEndpoint import BillingIdreportsEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import BillingReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class BillingreportsEndpoint( + SimpleSatEndpoint, + IGettable[BillingReports, SimpleSatRequestParams], + IPaginateable[BillingReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "billing_reports", parent_endpoint=parent_endpoint) + IGettable.__init__(self, BillingReports) + IPaginateable.__init__(self, BillingReports) + + def id(self, id: int) -> BillingIdreportsEndpoint: + """ + Sets the ID for this endpoint and returns an initialized BillingIdreportsEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + BillingIdreportsEndpoint: The initialized BillingIdreportsEndpoint object. + """ + child = BillingIdreportsEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[BillingReports]: + """ + Performs a GET request against the /billing_reports 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[BillingReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + BillingReports, + self, + "billing_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> BillingReports: + """ + Performs a GET request against the /billing_reports 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_many( + BillingReports, + super()._make_request("GET", data=data, params=params).json().get('billing_reports', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/BillingreportsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/BillingreportsIdEndpoint.py new file mode 100644 index 0000000..9fc9775 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/BillingreportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import BillingReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class BillingIdreportsEndpoint( + SimpleSatEndpoint, + IGettable[BillingReports, SimpleSatRequestParams], + IPaginateable[BillingReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, BillingReports) + IPaginateable.__init__(self, BillingReports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[BillingReports]: + """ + Performs a GET request against the /billing_reports 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[BillingReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + BillingReports, + self, + "billing_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> BillingReports: + """ + Performs a GET request against the /billing_reports 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( + BillingReports, + super()._make_request("GET", data=data, params=params).json().get('billing_report', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/IncidentreportsEndpoint.py b/src/pysimplesat/endpoints/simplesat/IncidentreportsEndpoint.py new file mode 100644 index 0000000..c15af96 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/IncidentreportsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.IncidentreportsIdEndpoint import IncidentreportsIdEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import IncidentReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class IncidentreportsEndpoint( + SimpleSatEndpoint, + IGettable[IncidentReports, SimpleSatRequestParams], + IPaginateable[IncidentReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "incident_reports", parent_endpoint=parent_endpoint) + IGettable.__init__(self, IncidentReports) + IPaginateable.__init__(self, IncidentReports) + + def id(self, id: int) -> IncidentreportsIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized IncidentreportsIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + IncidentreportsIdEndpoint: The initialized IncidentreportsIdEndpoint object. + """ + child = IncidentreportsIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[IncidentReports]: + """ + Performs a GET request against the /incident_reports 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[IncidentReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + IncidentReports, + self, + "incident_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> IncidentReports: + """ + Performs a GET request against the /incident_reports 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_many( + IncidentReports, + super()._make_request("GET", data=data, params=params).json().get('incident_reports', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/IncidentreportsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/IncidentreportsIdEndpoint.py new file mode 100644 index 0000000..5a3c3f0 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/IncidentreportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import IncidentReports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class IncidentreportsIdEndpoint( + SimpleSatEndpoint, + IGettable[IncidentReports, SimpleSatRequestParams], + IPaginateable[IncidentReports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, IncidentReports) + IPaginateable.__init__(self, IncidentReports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[IncidentReports]: + """ + Performs a GET request against the /incident_reports 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[IncidentReports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + IncidentReports, + self, + "incident_reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> IncidentReports: + """ + Performs a GET request against the /incident_reports 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( + IncidentReports, + super()._make_request("GET", data=data, params=params).json().get('incident_report', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/OrganizationsEndpoint.py b/src/pysimplesat/endpoints/simplesat/OrganizationsEndpoint.py new file mode 100644 index 0000000..643b00c --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/OrganizationsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.OrganizationsIdEndpoint import OrganizationsIdEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Organizations +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class OrganizationsEndpoint( + SimpleSatEndpoint, + IGettable[Organizations, SimpleSatRequestParams], + IPaginateable[Organizations, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "organizations", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Organizations) + IPaginateable.__init__(self, Organizations) + + def id(self, id: int) -> OrganizationsIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized OrganizationsIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + OrganizationsIdEndpoint: The initialized OrganizationsIdEndpoint object. + """ + child = OrganizationsIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Organizations]: + """ + Performs a GET request against the /organizations 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[Organizations]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Organizations, + self, + "organizations", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Organizations: + """ + Performs a GET request against the /Organizations 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_many( + Organizations, + super()._make_request("GET", data=data, params=params).json().get('organizations', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/OrganizationsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/OrganizationsIdEndpoint.py new file mode 100644 index 0000000..19b480a --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/OrganizationsIdEndpoint.py @@ -0,0 +1,37 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, +) +from pysimplesat.models.simplesat import Organizations +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class OrganizationsIdEndpoint( + SimpleSatEndpoint, + IGettable[Organizations, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Organizations) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Organizations: + """ + Performs a GET request against the /organizations/{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( + Organizations, + super()._make_request("GET", data=data, params=params).json().get('organization', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/ReportsEndpoint.py b/src/pysimplesat/endpoints/simplesat/ReportsEndpoint.py new file mode 100644 index 0000000..d0b4439 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/ReportsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.ReportsIdEndpoint import ReportsIdEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Reports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class ReportsEndpoint( + SimpleSatEndpoint, + IGettable[Reports, SimpleSatRequestParams], + IPaginateable[Reports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "reports", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Reports) + IPaginateable.__init__(self, Reports) + + def id(self, id: int) -> ReportsIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized ReportsIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + ReportsIdEndpoint: The initialized ReportsIdEndpoint object. + """ + child = ReportsIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Reports]: + """ + Performs a GET request against the /reports 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[Reports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Reports, + self, + "reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Reports: + """ + Performs a GET request against the /reports 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_many( + Reports, + super()._make_request("GET", data=data, params=params).json().get('reports', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/ReportsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/ReportsIdEndpoint.py new file mode 100644 index 0000000..97f59da --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/ReportsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Reports +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class ReportsIdEndpoint( + SimpleSatEndpoint, + IGettable[Reports, SimpleSatRequestParams], + IPaginateable[Reports, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Reports) + IPaginateable.__init__(self, Reports) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Reports]: + """ + Performs a GET request against the /reports 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[Reports]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Reports, + self, + "reports", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Reports: + """ + Performs a GET request against the /reports 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( + Reports, + super()._make_request("GET", data=data, params=params).json().get('report', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/SignalsEndpoint.py b/src/pysimplesat/endpoints/simplesat/SignalsEndpoint.py new file mode 100644 index 0000000..125bae9 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/SignalsEndpoint.py @@ -0,0 +1,86 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.endpoints.simplesat.SignalsIdEndpoint import SignalsIdEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Signals +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class SignalsEndpoint( + SimpleSatEndpoint, + IGettable[Signals, SimpleSatRequestParams], + IPaginateable[Signals, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "signals", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Signals) + IPaginateable.__init__(self, Signals) + + def id(self, id: int) -> SignalsIdEndpoint: + """ + Sets the ID for this endpoint and returns an initialized SignalsIdEndpoint object to move down the chain. + + Parameters: + id (int): The ID to set. + Returns: + SignalsIdEndpoint: The initialized SignalsIdEndpoint object. + """ + child = SignalsIdEndpoint(self.client, parent_endpoint=self) + child._id = id + return child + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Signals]: + """ + Performs a GET request against the /signals 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[Signals]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Signals, + self, + "signals", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Signals: + """ + Performs a GET request against the /signals 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_many( + Signals, + super()._make_request("GET", data=data, params=params).json().get('signals', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/SignalsIdEndpoint.py b/src/pysimplesat/endpoints/simplesat/SignalsIdEndpoint.py new file mode 100644 index 0000000..e3075e7 --- /dev/null +++ b/src/pysimplesat/endpoints/simplesat/SignalsIdEndpoint.py @@ -0,0 +1,72 @@ +from pysimplesat.endpoints.base.base_endpoint import SimpleSatEndpoint +from pysimplesat.interfaces import ( + IGettable, + IPaginateable, +) +from pysimplesat.models.simplesat import Signals +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, +) + + +class SignalsIdEndpoint( + SimpleSatEndpoint, + IGettable[Signals, SimpleSatRequestParams], + IPaginateable[Signals, SimpleSatRequestParams], +): + def __init__(self, client, parent_endpoint=None) -> None: + SimpleSatEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint) + IGettable.__init__(self, Signals) + IPaginateable.__init__(self, Signals) + + def paginated( + self, + page: int, + limit: int, + params: SimpleSatRequestParams | None = None, + ) -> PaginatedResponse[Signals]: + """ + Performs a GET request against the /signals 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[Signals]: The initialized PaginatedResponse object. + """ + if params: + params["page"] = page + params["limit"] = limit + else: + params = {"page": page, "limit": limit} + return PaginatedResponse( + super()._make_request("GET", params=params), + Signals, + self, + "signals", + page, + limit, + params, + ) + + def get( + self, + data: JSON | None = None, + params: SimpleSatRequestParams | None = None, + ) -> Signals: + """ + Performs a GET request against the /signals 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( + Signals, + super()._make_request("GET", data=data, params=params).json().get('signal', {}), + ) diff --git a/src/pysimplesat/endpoints/simplesat/__init__.py b/src/pysimplesat/endpoints/simplesat/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/exceptions.py b/src/pysimplesat/exceptions.py new file mode 100644 index 0000000..065bbcf --- /dev/null +++ b/src/pysimplesat/exceptions.py @@ -0,0 +1,89 @@ +import json +from typing import ClassVar +from urllib.parse import urlsplit, urlunsplit + +from requests import JSONDecodeError, Response + + +class SimpleSatException(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(SimpleSatException): + _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(SimpleSatException): + _code_explanation = "Unauthorized" + _error_suggestion = "Please check your credentials are correct before retrying." + + +class PermissionsFailedException(SimpleSatException): + _code_explanation = "Forbidden" + _error_suggestion = "You may be attempting to access a resource you do not have the appropriate permissions for." + + +class NotFoundException(SimpleSatException): + _code_explanation = "Not Found" + _error_suggestion = "You may be attempting to access a resource that has been moved or deleted." + + +class MethodNotAllowedException(SimpleSatException): + _code_explanation = "Method Not Allowed" + _error_suggestion = "This resource does not support the HTTP method you are trying to use." + + +class ConflictException(SimpleSatException): + _code_explanation = "Conflict" + _error_suggestion = "This resource is possibly in use or conflicts with another record." + +class TooManyRequestsException(SimpleSatException): + _code_explanation = "Too Many Requests" + _error_suggestion = "This resource is currently being rate limited. Please wait and try again." + + +class ServerError(SimpleSatException): + _code_explanation = "Internal Server Error" + + +class ObjectExistsError(SimpleSatException): + _code_explanation = "Object Exists" + _error_suggestion = "This resource already exists." diff --git a/src/pysimplesat/interfaces.py b/src/pysimplesat/interfaces.py new file mode 100644 index 0000000..f78b6f2 --- /dev/null +++ b/src/pysimplesat/interfaces.py @@ -0,0 +1,102 @@ +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Generic, TypeVar + +from pysimplesat.responses.paginated_response import PaginatedResponse +from pysimplesat.types import ( + JSON, + SimpleSatRequestParams, + PatchRequestData, +) + +if TYPE_CHECKING: + from pydantic import BaseModel + +TModel = TypeVar("TModel", bound="BaseModel") +TRequestParams = TypeVar( + "TRequestParams", + bound=SimpleSatRequestParams, +) + + +class IMethodBase(ABC, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + self.model = model + + +class IPaginateable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def paginated( + self, + page: int, + page_size: int, + params: TRequestParams | None = None, + ) -> PaginatedResponse[TModel]: + pass + + +class IGettable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def get( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPostable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def post( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPatchable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def patch( + self, + data: PatchRequestData, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IPuttable(IMethodBase, Generic[TModel, TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def put( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> TModel: + pass + + +class IDeleteable(IMethodBase, Generic[TRequestParams]): + def __init__(self, model: TModel) -> None: + super().__init__(model) + + @abstractmethod + def delete( + self, + data: JSON | None = None, + params: TRequestParams | None = None, + ) -> None: + pass diff --git a/src/pysimplesat/models/__init__.py b/src/pysimplesat/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/models/base/__init__.py b/src/pysimplesat/models/base/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/models/base/base_model.py b/src/pysimplesat/models/base/base_model.py new file mode 100644 index 0000000..4486ba7 --- /dev/null +++ b/src/pysimplesat/models/base/base_model.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import inspect +from types import UnionType +from typing import Union, get_args, get_origin + +from pydantic import BaseModel, ConfigDict + +from pysimplesat.utils.naming import to_camel_case + + +class SimpleSatModel(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, SimpleSatModel): + 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, SimpleSatModel): + 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, SimpleSatModel): + was_model = True + for sk, sv in v.annotation._get_field_names_and_types().items(): + field_names_and_types[f"{v.alias}/{sk}"] = sv + elif v.annotation is not None and v.annotation.__name__ != "NoneType": + field_type = v.annotation.__name__ + + if not was_model: + field_names_and_types[v.alias] = field_type + + return field_names_and_types diff --git a/src/pysimplesat/models/base/message_model.py b/src/pysimplesat/models/base/message_model.py new file mode 100644 index 0000000..5bd7a02 --- /dev/null +++ b/src/pysimplesat/models/base/message_model.py @@ -0,0 +1,5 @@ +from pydantic import BaseModel + + +class GenericMessageModel(BaseModel): + message: str diff --git a/src/pysimplesat/models/simplesat/__init__.py b/src/pysimplesat/models/simplesat/__init__.py new file mode 100644 index 0000000..ada6e18 --- /dev/null +++ b/src/pysimplesat/models/simplesat/__init__.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import Field + +from pysimplesat.models.base.base_model import SimpleSatModel + +class Pagination(SimpleSatModel): + 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(SimpleSatModel): + 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 Answer(SimpleSatModel): + 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(SimpleSatModel): + 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(SimpleSatModel): + survey_id: int | None = Field(default=None, alias="SurveyId") + tags: list | None = Field(default=None, alias="Tags") + answers: dict[str, Any] | None = Field(default=None, alias="Answers") + team_members: 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") diff --git a/src/pysimplesat/py.typed b/src/pysimplesat/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/responses/__init__.py b/src/pysimplesat/responses/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/responses/paginated_response.py b/src/pysimplesat/responses/paginated_response.py new file mode 100644 index 0000000..ecc7347 --- /dev/null +++ b/src/pysimplesat/responses/paginated_response.py @@ -0,0 +1,204 @@ +from __future__ import annotations +import json + +from typing import TYPE_CHECKING, Generic, TypeVar + +from pysimplesat.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 pysimplesat.types import RequestParams + + +TModel = TypeVar("TModel", bound="BaseModel") + +if TYPE_CHECKING: + from pysimplesat.interfaces import IPaginateable + + +class PaginatedResponse(Generic[TModel]): + """ + PaginatedResponse is a wrapper class for handling paginated responses from the + SimpleSat 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 SimpleSatEndpoint 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 + SimpleSat 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 SimpleSatEndpoint 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 (SimpleSatEndpoint[TModel]): The endpointmodel associated with the response. + endpoint: The endpoint url to extract the data + limit (int): The number of items per page. + """ + self.response = response + self.response_model = response_model + self.endpointmodel = endpointmodel + self.endpoint = endpoint + self.limit = limit + # Get page data from the response body + try: + self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).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: + # SimpleSat API gives us a handy response to parse for Pagination + self.has_next_page: bool = self.parsed_pagination_response.get("has_next_page", False) + self.has_prev_page: bool = self.parsed_pagination_response.get("has_prev_page", False) + self.first_page: int = self.parsed_pagination_response.get("first_page", None) + self.prev_page: int = self.parsed_pagination_response.get("prev_page", None) + self.next_page: int = self.parsed_pagination_response.get("next_page", None) + self.last_page: int = self.parsed_pagination_response.get("last_page", None) + else: + # Haven't worked on this yet + self.has_next_page: bool = True + self.has_prev_page: bool = page > 1 + self.first_page: int = 1 + self.prev_page = page - 1 if page > 1 else 1 + self.next_page = page + 1 + self.last_page = 999999 + self.data: list[TModel] = [response_model.model_validate(d) for d in response.json().get(endpoint, {})] + self.has_data = self.data and len(self.data) > 0 + self.index = 0 + + def get_next_page(self) -> PaginatedResponse[TModel]: + """ + Fetch the next page of the paginated response. + + Returns: + PaginatedResponse[TModel]: The updated PaginatedResponse instance + with the data from the next page or None if there is no next page. + """ + if not self.has_next_page or not self.next_page: + self.has_data = False + return self + + next_response = self.endpointmodel.paginated(self.next_page, self.limit, self.params) + self._initialize( + next_response.response, + next_response.response_model, + next_response.endpointmodel, + next_response.endpoint, + self.next_page, + next_response.limit, + self.params, + ) + return self + + def get_previous_page(self) -> PaginatedResponse[TModel]: + """ + Fetch the next page of the paginated response. + + Returns: + PaginatedResponse[TModel]: The updated PaginatedResponse instance + with the data from the next page or None if there is no next page. + """ + if not self.has_prev_page or not self.prev_page: + self.has_data = False + return self + + prev_response = self.endpointmodel.paginated(self.prev_page, self.limit, self.params) + self._initialize( + prev_response.response, + prev_response.response_model, + prev_response.endpointmodel, + self.prev_page, + prev_response.limit, + self.params, + ) + return self + + def all(self) -> Iterable[TModel]: + """ + Iterate through all items in the paginated response, across all pages. + + Yields: + TModel: An instance of the model class for each item in the paginated response. + """ + while self.has_data: + yield from self.data + self.get_next_page() + + def __iter__(self): + """ + Implement the iterator protocol for the PaginatedResponse class. + + Returns: + PaginatedResponse[TModel]: The current instance of the PaginatedResponse. + """ + return self + + def __dict__(self): + """ + Implement the iterator protocol for the PaginatedResponse class. + + Returns: + PaginatedResponse[TModel]: The current instance of the PaginatedResponse. + """ + return self.data + + def __next__(self): + """ + Implement the iterator protocol by getting the next item in the data. + + Returns: + TModel: The next item in the data. + + Raises: + StopIteration: If there are no more items in the data. + """ + if self.index < len(self.data): + result = self.data[self.index] + self.index += 1 + return result + else: + raise StopIteration diff --git a/src/pysimplesat/types.py b/src/pysimplesat/types.py new file mode 100644 index 0000000..62b766c --- /dev/null +++ b/src/pysimplesat/types.py @@ -0,0 +1,42 @@ +from typing import Literal, TypeAlias + +from typing_extensions import NotRequired, TypedDict +from datetime import datetime + +Literals: TypeAlias = str | int | float | bool +JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | Literals | None + + +class Patch(TypedDict): + op: Literal["add"] | Literal["replace"] | Literal["remove"] + path: str + value: JSON + + +class SimpleSatRequestParams(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 = SimpleSatRequestParams | GenericRequestParams +PatchRequestData: TypeAlias = list[Patch] +RequestData: TypeAlias = JSON | PatchRequestData +RequestMethod: TypeAlias = Literal["GET", "POST", "PUT", "PATCH", "DELETE"] diff --git a/src/pysimplesat/utils/__init__.py b/src/pysimplesat/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/utils/experimental/__init__.py b/src/pysimplesat/utils/experimental/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pysimplesat/utils/experimental/condition.py b/src/pysimplesat/utils/experimental/condition.py new file mode 100644 index 0000000..23b1335 --- /dev/null +++ b/src/pysimplesat/utils/experimental/condition.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import inspect +import re +from datetime import datetime +from enum import Enum +from typing import TYPE_CHECKING, Any, Generic, TypeVar + +from pysimplesat.utils.naming import to_camel_case + +if TYPE_CHECKING: + from collections.abc import Callable + +T = TypeVar("T") + + +class ValueType(Enum): + STR = 1 + INT = 2 + DATETIME = 3 + + +class Condition(Generic[T]): + def __init__(self: Condition[T]) -> None: + self._condition_string: str = "" + self._field = "" + + def field(self: Condition[T], selector: Callable[[type[T]], Any]) -> Condition[T]: + field = "" + + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"field\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = to_camel_case("/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:])) + + finally: + del frame + + self._condition_string += field + return self + + def equals(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " = " + self.__add_typed_value_to_string(value, type(value)) + return self + + def not_equals(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " = " + self.__add_typed_value_to_string(value, type(value)) + return self + + def less_than(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " < " + self.__add_typed_value_to_string(value, type(value)) + return self + + def less_than_or_equals( + self: Condition[T], + value: Any, # noqa: ANN401 + ) -> Condition[T]: + self._condition_string += " <= " + self.__add_typed_value_to_string(value, type(value)) + return self + + def greater_than(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " > " + self.__add_typed_value_to_string(value, type(value)) + return self + + def greater_than_or_equals( + self: Condition[T], + value: Any, # noqa: ANN401 + ) -> Condition[T]: + self._condition_string += " >= " + self.__add_typed_value_to_string(value, type(value)) + return self + + def contains(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " CONTAINS " + self.__add_typed_value_to_string(value, type(value)) + return self + + def like(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " LIKE " + self.__add_typed_value_to_string(value, type(value)) + return self + + def in_(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " IN " + self.__add_typed_value_to_string(value, type(value)) + return self + + def not_(self: Condition[T], value: Any) -> Condition[T]: # noqa: ANN401 + self._condition_string += " NOT " + self.__add_typed_value_to_string(value, type(value)) + return self + + def __add_typed_value_to_string( # noqa: ANN202 + self: Condition[T], + value: Any, # noqa: ANN401 + type: type, # noqa: A002 + ): + if type is str: + self._condition_string += f'"{value}"' + elif type is int: # noqa: SIM114 + self._condition_string += str(value) + elif type is bool: + self._condition_string += str(value) + elif type is datetime: + self._condition_string += f"[{value}]" + else: + self._condition_string += f'"{value}"' + + def and_(self: Condition[T], selector: Callable[[type[T]], Any] | None = None) -> Condition[T]: + self._condition_string += " AND " + + if selector is not None: + field = "" + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"and_\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = "/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:]) + + finally: + del frame + + self._condition_string += field + return self + + def or_(self: Condition[T], selector: Callable[[type[T]], Any] | None = None) -> Condition[T]: + self._condition_string += " OR " + + if selector is not None: + field = "" + frame = inspect.currentframe() + try: + context = inspect.getframeinfo(frame.f_back).code_context + caller_lines = "".join([line.strip() for line in context]) + m = re.search(r"or_\s*\(([^)]+)\)", caller_lines) + if m: + caller_lines = m.group(1) + + field = "/".join(caller_lines.replace("(", "").replace(")", "").split(".")[1:]) + + finally: + del frame + + self._condition_string += field + return self + + def wrap(self: Condition[T], condition: Callable[[Condition[T]], Condition[T]]) -> Condition[T]: + self._condition_string += f"({condition(Condition[T]())})" + return self + + def __str__(self: Condition[T]) -> str: + return self._condition_string.strip() diff --git a/src/pysimplesat/utils/experimental/patch_maker.py b/src/pysimplesat/utils/experimental/patch_maker.py new file mode 100644 index 0000000..55f83d6 --- /dev/null +++ b/src/pysimplesat/utils/experimental/patch_maker.py @@ -0,0 +1,37 @@ +import json +from enum import Enum +from typing import Any + + +class Patch: + class PatchOp(Enum): + """ + PatchOperation is an enumeration of the different patch operations supported + by the SimpleSat API. These operations are ADD, REPLACE, and REMOVE. + """ + + ADD = 1 + REPLACE = 2 + REMOVE = 3 + + def __init__(self, op: PatchOp, path: str, value: Any) -> None: # noqa: ANN401 + self.op = op.name.lower() + self.path = path + self.value = value + + def __repr__(self) -> str: + """ + Return a string representation of the model as a formatted JSON string. + + Returns: + str: A formatted JSON string representation of the model. + """ + return json.dumps(self.__dict__, default=str, indent=2) + + +class PatchGroup: + def __init__(self, *patches: Patch) -> None: + self.patches = list(patches) + + def __repr__(self) -> str: + return str(self.patches) diff --git a/src/pysimplesat/utils/helpers.py b/src/pysimplesat/utils/helpers.py new file mode 100644 index 0000000..e258dd3 --- /dev/null +++ b/src/pysimplesat/utils/helpers.py @@ -0,0 +1,190 @@ +import re +import math +from datetime import datetime +from typing import Any + +from requests.structures import CaseInsensitiveDict + + +def cw_format_datetime(dt: datetime) -> str: + """Format a datetime object as a string in ISO 8601 format. This is the format that SimpleSat uses. + + Args: + dt (datetime): The datetime object to be formatted. + + Returns: + str: The formatted datetime string in the format "YYYY-MM-DDTHH:MM:SSZ". + + Example: + from datetime import datetime + + dt = datetime(2022, 1, 1, 12, 0, 0) + formatted_dt = cw_format_datetime(dt) + print(formatted_dt) # Output: "2022-01-01T12:00:00Z" + """ + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + +def parse_response_body( + body: CaseInsensitiveDict, +) -> dict[str, Any] | None: + """ + Parses response body to extract pagination information. + + Arguments: + - body: content.json().get('pagination', {}) A dictionary containing the headers of an HTTP response. + + Returns: + - A dictionary containing the extracted pagination information. The keys in the dictionary include: + - "first_page": An optional integer representing the number of the first page. + - "prev_page": An optional integer representing the number of the previous page. + - "next_page": An optional integer representing the number of the next page. + - "last_page": An optional integer representing the number of the last page. + - "has_next_page": A boolean indicating whether there is a next page. + - "has_prev_page": A boolean indicating whether there is a previous page. + + If the "Link" header is not present in the headers dictionary, None is returned. + + Example Usage: + headers = { + "Link": '; rel="first", ; rel="next"' + } + pagination_info = parse_link_headers(headers) + print(pagination_info) + # Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True} + """ + if body.get("current_page") is None: + return None + has_next_page: bool = False + has_prev_page: bool = False + first_page: int | None = None + prev_page: int | None = None + current_page: int | None = None + current_page_count: int | None = None + limit: int | None = None + total_count: int | None = None + next_page: int | None = None + next_page_url: str | None = None + next_page_token: str | None = None + last_page: int | None = None + + result = {} + + if body.get("first_page") is not None: + result["first_page"] = body.get("first_page") + + if body.get("prev_page") is not None: + result["prev_page"] = body.get("prev_page") + elif body.get("current_page") is not None: + if body.get("current_page") > 1: + result["prev_page"] = body.get("current_page") - 1 + elif body.get("currentPage") is not None: + if body.get("currentPage") > 1: + result["prev_page"] = body.get("currentPage") - 1 + + if body.get("next_page") is not None: + result["next_page"] = body.get("next_page") + elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"): + result["next_page"] = body.get("currentPage") + 1 + + if body.get("last_page") is not None: + result["last_page"] = body.get("last_page") + elif body.get("lastPage") is not None: + result["last_page"] = body.get("lastPage") + elif body.get("last_page") is None and body.get("current_page") is not None: + result["last_page"] = math.ceil(body.get("total_count")/body.get("limit")) + + if body.get("has_next_page"): + result["has_next_page"] = body.get("has_next_page") + elif body.get("current_page") is not None and body.get("next_page") is not None: + result["has_next_page"] = True + elif body.get("current_page") is not None and body.get("next_page") is None: + result["has_next_page"] = False + elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"): + result["has_next_page"] = True + + if body.get("has_prev_page"): + result["has_prev_page"] = body.get("has_prev_page") + elif body.get("current_page") is not None: + if body.get("current_page") > 1: + result["has_prev_page"] = True + elif body.get("currentPage") is not None: + if body.get("currentPage") > 1: + result["has_prev_page"] = True + + return result + +def parse_link_headers( + headers: CaseInsensitiveDict, +) -> dict[str, Any] | None: + """ + Parses link headers to extract pagination information. + + Arguments: + - headers: A dictionary containing the headers of an HTTP response. The value associated with the "Link" key should be a string representing the link headers. + + Returns: + - A dictionary containing the extracted pagination information. The keys in the dictionary include: + - "first_page": An optional integer representing the number of the first page. + - "prev_page": An optional integer representing the number of the previous page. + - "next_page": An optional integer representing the number of the next page. + - "last_page": An optional integer representing the number of the last page. + - "has_next_page": A boolean indicating whether there is a next page. + - "has_prev_page": A boolean indicating whether there is a previous page. + + If the "Link" header is not present in the headers dictionary, None is returned. + + Example Usage: + headers = { + "Link": '; rel="first", ; rel="next"' + } + pagination_info = parse_link_headers(headers) + print(pagination_info) + # Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True} + """ + if headers.get("Link") is None: + return None + links = headers["Link"].split(",") + has_next_page: bool = False + has_prev_page: bool = False + first_page: int | None = None + prev_page: int | None = None + next_page: int | None = None + last_page: int | None = None + + for link in links: + match = re.search(r'page=(\d+)>; rel="(.*?)"', link) + if match: + page_number = int(match.group(1)) + rel_value = match.group(2) + if rel_value == "first": + first_page = page_number + elif rel_value == "prev": + prev_page = page_number + has_prev_page = True + elif rel_value == "next": + next_page = page_number + has_next_page = True + elif rel_value == "last": + last_page = page_number + + result = {} + + if first_page is not None: + result["first_page"] = first_page + + if prev_page is not None: + result["prev_page"] = prev_page + + if next_page is not None: + result["next_page"] = next_page + + if last_page is not None: + result["last_page"] = last_page + + if has_next_page: + result["has_next_page"] = has_next_page + + if has_prev_page: + result["has_prev_page"] = has_prev_page + + return result diff --git a/src/pysimplesat/utils/naming.py b/src/pysimplesat/utils/naming.py new file mode 100644 index 0000000..e0d1758 --- /dev/null +++ b/src/pysimplesat/utils/naming.py @@ -0,0 +1,23 @@ +from keyword import iskeyword + + +def to_snake_case(string: str) -> str: + return ("_" if string.startswith("_") else "") + "".join( + ["_" + i.lower() if i.isupper() else i for i in string.lstrip("_")] + ).lstrip("_") + + +def to_camel_case(string: str) -> str: + string_split = string.split("_") + return string_split[0] + "".join(word.capitalize() for word in string_split[1:]) + + +def to_title_case_preserve_case(string: str) -> str: + return string[:1].upper() + string[1:] + + +def ensure_not_reserved(string: str) -> str: + if iskeyword(string): + return string + "_" + else: # noqa: RET505 + return string diff --git a/src/simplesat_scratchpad.py b/src/simplesat_scratchpad.py new file mode 100644 index 0000000..4fab02b --- /dev/null +++ b/src/simplesat_scratchpad.py @@ -0,0 +1,15 @@ +import os +from pysimplesat import SimpleSatAPIClient +from dotenv import load_dotenv + +load_dotenv() + +privatekey = os.getenv('SIMPLESAT_API_TOKEN') + +# init client +simplesat_api_client = SimpleSatAPIClient( + privatekey, +) + +#account = simplesat_api_client.account.get() +#print(account)