Datapath/pysimplesat
2
0
Fork 0
mirror of https://github.com/brygphilomena/pysimplesat.git synced 2025-11-04 00:47:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Prep with pyhuntress base code

Browse Source
This commit is contained in:
Peter Annabel 2025-07-30 15:58:55 -05:00
parent d97f02a96b
commit 56ce6105ea
51 changed files with 2910 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
pyproject.toml Normal file
View File

@ -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"]

3
requirements.txt Normal file
View File

@ -0,0 +1,3 @@
requests==2.32.4
pydantic==2.11.7
typing_extensions==4.14.1

4
src/pysimplesat/__init__.py Normal file
View File

@ -0,0 +1,4 @@
from pysimplesat.clients.simplesat_client import SimpleSatAPIClient
__all__ = ["SimpleSatAPIClient"]
__version__ = "0.1.1"

0
src/pysimplesat/clients/__init__.py Normal file
View File

134
src/pysimplesat/clients/base_client.py Normal file
View File

@ -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

106
src/pysimplesat/clients/simplesat_client.py Normal file
View File

@ -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}",
}

9
src/pysimplesat/config.py Normal file
View File

@ -0,0 +1,9 @@
class Config:
def __init__(self, max_retries=3) -> None:
"""
Initializes a new instance of the Config class.
Args:
max_retries (int): The maximum number of retries for a retryable HTTP operation (500) (default = 3)
"""
self.max_retries = max_retries

0
src/pysimplesat/endpoints/__init__.py Normal file
View File

0
src/pysimplesat/endpoints/base/__init__.py Normal file
View File

163
src/pysimplesat/endpoints/base/base_endpoint.py Normal file
View File

@ -0,0 +1,163 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, TypeVar
if TYPE_CHECKING:
from pydantic import BaseModel
from requests import Response
from 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)

37
src/pysimplesat/endpoints/siem/AccountEndpoint.py Normal file
View File

@ -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', {}),
)

37
src/pysimplesat/endpoints/siem/ActorEndpoint.py Normal file
View File

@ -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(),
)

72
src/pysimplesat/endpoints/siem/AgentsIdEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/siem/BillingreportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/siem/IncidentreportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

37
src/pysimplesat/endpoints/siem/OrganizationsIdEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/siem/ReportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/siem/SignalsIdEndpoint.py Normal file
View File

@ -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', {}),
)

37
src/pysimplesat/endpoints/simplesat/AccountEndpoint.py Normal file
View File

@ -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', {}),
)

37
src/pysimplesat/endpoints/simplesat/ActorEndpoint.py Normal file
View File

@ -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(),
)

86
src/pysimplesat/endpoints/simplesat/AgentsEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/simplesat/AgentsIdEndpoint.py Normal file
View File

@ -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', {}),
)

86
src/pysimplesat/endpoints/simplesat/BillingreportsEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/simplesat/BillingreportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

86
src/pysimplesat/endpoints/simplesat/IncidentreportsEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/simplesat/IncidentreportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

86
src/pysimplesat/endpoints/simplesat/OrganizationsEndpoint.py Normal file
View File

@ -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', {}),
)

37
src/pysimplesat/endpoints/simplesat/OrganizationsIdEndpoint.py Normal file
View File

@ -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', {}),
)

86
src/pysimplesat/endpoints/simplesat/ReportsEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/simplesat/ReportsIdEndpoint.py Normal file
View File

@ -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', {}),
)

86
src/pysimplesat/endpoints/simplesat/SignalsEndpoint.py Normal file
View File

@ -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', {}),
)

72
src/pysimplesat/endpoints/simplesat/SignalsIdEndpoint.py Normal file
View File

@ -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', {}),
)

0
src/pysimplesat/endpoints/simplesat/__init__.py Normal file
View File

89
src/pysimplesat/exceptions.py Normal file
View File

@ -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."

102
src/pysimplesat/interfaces.py Normal file
View File

@ -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

0
src/pysimplesat/models/__init__.py Normal file
View File

0
src/pysimplesat/models/base/__init__.py Normal file
View File

60
src/pysimplesat/models/base/base_model.py Normal file
View File

@ -0,0 +1,60 @@
from __future__ import annotations
import inspect
from types import UnionType
from typing import Union, get_args, get_origin
from pydantic import BaseModel, ConfigDict
from 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

5
src/pysimplesat/models/base/message_model.py Normal file
View File

@ -0,0 +1,5 @@
from pydantic import BaseModel
class GenericMessageModel(BaseModel):
message: str

61
src/pysimplesat/models/simplesat/__init__.py Normal file
View File

@ -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")

0
src/pysimplesat/py.typed Normal file
View File

0
src/pysimplesat/responses/__init__.py Normal file
View File

204
src/pysimplesat/responses/paginated_response.py Normal file
View File

@ -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

42
src/pysimplesat/types.py Normal file
View File

@ -0,0 +1,42 @@
from typing import Literal, TypeAlias
from typing_extensions import NotRequired, TypedDict
from datetime import datetime
Literals: TypeAlias = str | int | float | bool
JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | Literals | None
class Patch(TypedDict):
op: Literal["add"] | Literal["replace"] | Literal["remove"]
path: str
value: JSON
class 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"]

0
src/pysimplesat/utils/__init__.py Normal file
View File

0
src/pysimplesat/utils/experimental/__init__.py Normal file
View File

166
src/pysimplesat/utils/experimental/condition.py Normal file
View File

@ -0,0 +1,166 @@
from __future__ import annotations
import inspect
import re
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING, Any, Generic, TypeVar
from 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()

37
src/pysimplesat/utils/experimental/patch_maker.py Normal file
View File

@ -0,0 +1,37 @@
import json
from enum import Enum
from typing import Any
class Patch:
class PatchOp(Enum):
"""
PatchOperation is an enumeration of the different patch operations supported
by the 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)

190
src/pysimplesat/utils/helpers.py Normal file
View File

@ -0,0 +1,190 @@
import re
import math
from datetime import datetime
from typing import Any
from requests.structures import CaseInsensitiveDict
def cw_format_datetime(dt: datetime) -> str:
"""Format a datetime object as a string in ISO 8601 format. This is the format that 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": '<https://example.com/api?page=1>; rel="first", <https://example.com/api?page=2>; rel="next"'
}
pagination_info = parse_link_headers(headers)
print(pagination_info)
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
"""
if body.get("current_page") is None:
return None
has_next_page: bool = False
has_prev_page: bool = False
first_page: int | None = None
prev_page: int | None = None
current_page: int | None = None
current_page_count: int | None = None
limit: int | None = None
total_count: int | None = None
next_page: int | None = None
next_page_url: str | None = None
next_page_token: str | None = None
last_page: int | None = None
result = {}
if body.get("first_page") is not None:
result["first_page"] = body.get("first_page")
if body.get("prev_page") is not None:
result["prev_page"] = body.get("prev_page")
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["prev_page"] = body.get("current_page") - 1
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["prev_page"] = body.get("currentPage") - 1
if body.get("next_page") is not None:
result["next_page"] = body.get("next_page")
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["next_page"] = body.get("currentPage") + 1
if body.get("last_page") is not None:
result["last_page"] = body.get("last_page")
elif body.get("lastPage") is not None:
result["last_page"] = body.get("lastPage")
elif body.get("last_page") is None and body.get("current_page") is not None:
result["last_page"] = math.ceil(body.get("total_count")/body.get("limit"))
if body.get("has_next_page"):
result["has_next_page"] = body.get("has_next_page")
elif body.get("current_page") is not None and body.get("next_page") is not None:
result["has_next_page"] = True
elif body.get("current_page") is not None and body.get("next_page") is None:
result["has_next_page"] = False
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["has_next_page"] = True
if body.get("has_prev_page"):
result["has_prev_page"] = body.get("has_prev_page")
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["has_prev_page"] = True
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["has_prev_page"] = True
return result
def parse_link_headers(
headers: CaseInsensitiveDict,
) -> dict[str, Any] | None:
"""
Parses link headers to extract pagination information.
Arguments:
- headers: A dictionary containing the headers of an HTTP response. The value associated with the "Link" key should be a string representing the link headers.
Returns:
- A dictionary containing the extracted pagination information. The keys in the dictionary include:
- "first_page": An optional integer representing the number of the first page.
- "prev_page": An optional integer representing the number of the previous page.
- "next_page": An optional integer representing the number of the next page.
- "last_page": An optional integer representing the number of the last page.
- "has_next_page": A boolean indicating whether there is a next page.
- "has_prev_page": A boolean indicating whether there is a previous page.
If the "Link" header is not present in the headers dictionary, None is returned.
Example Usage:
headers = {
"Link": '<https://example.com/api?page=1>; rel="first", <https://example.com/api?page=2>; rel="next"'
}
pagination_info = parse_link_headers(headers)
print(pagination_info)
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
"""
if headers.get("Link") is None:
return None
links = headers["Link"].split(",")
has_next_page: bool = False
has_prev_page: bool = False
first_page: int | None = None
prev_page: int | None = None
next_page: int | None = None
last_page: int | None = None
for link in links:
match = re.search(r'page=(\d+)>; rel="(.*?)"', link)
if match:
page_number = int(match.group(1))
rel_value = match.group(2)
if rel_value == "first":
first_page = page_number
elif rel_value == "prev":
prev_page = page_number
has_prev_page = True
elif rel_value == "next":
next_page = page_number
has_next_page = True
elif rel_value == "last":
last_page = page_number
result = {}
if first_page is not None:
result["first_page"] = first_page
if prev_page is not None:
result["prev_page"] = prev_page
if next_page is not None:
result["next_page"] = next_page
if last_page is not None:
result["last_page"] = last_page
if has_next_page:
result["has_next_page"] = has_next_page
if has_prev_page:
result["has_prev_page"] = has_prev_page
return result

23
src/pysimplesat/utils/naming.py Normal file
View File

@ -0,0 +1,23 @@
from keyword import iskeyword
def to_snake_case(string: str) -> str:
return ("_" if string.startswith("_") else "") + "".join(
["_" + i.lower() if i.isupper() else i for i in string.lstrip("_")]
).lstrip("_")
def to_camel_case(string: str) -> str:
string_split = string.split("_")
return string_split[0] + "".join(word.capitalize() for word in string_split[1:])
def to_title_case_preserve_case(string: str) -> str:
return string[:1].upper() + string[1:]
def ensure_not_reserved(string: str) -> str:
if iskeyword(string):
return string + "_"
else: # noqa: RET505
return string

15
src/simplesat_scratchpad.py Normal file
View File

@ -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)