Added pagination and missing endpoints in ironscales client

This commit is contained in:
Peter Annabel 2025-08-04 21:24:57 -05:00
parent 00aa2fd4fd
commit c45bf96447
15 changed files with 449 additions and 68 deletions

View File

@ -11,6 +11,10 @@ if typing.TYPE_CHECKING:
from pyironscales.endpoints.ironscales.EmailsEndpoint import EmailsEndpoint
from pyironscales.endpoints.ironscales.IncidentEndpoint import IncidentEndpoint
from pyironscales.endpoints.ironscales.IntegrationsEndpoint import IntegrationsEndpoint
from pyironscales.endpoints.ironscales.MailboxesEndpoint import MailboxesEndpoint
from pyironscales.endpoints.ironscales.MitigationEndpoint import MitigationEndpoint
from pyironscales.endpoints.ironscales.PlanDetailsEndpoint import PlanDetailsEndpoint
from pyironscales.endpoints.ironscales.SettingsEndpoint import SettingsEndpoint
class IronscalesAPIClient(IronscalesClient):
@ -68,6 +72,30 @@ class IronscalesAPIClient(IronscalesClient):
return IntegrationsEndpoint(self)
@property
def mailboxes(self) -> "MailboxesEndpoint":
from pyironscales.endpoints.ironscales.MailboxesEndpoint import MailboxesEndpoint
return MailboxesEndpoint(self)
@property
def mitigation(self) -> "MitigationEndpoint":
from pyironscales.endpoints.ironscales.MitigationEndpoint import MitigationEndpoint
return MitigationEndpoint(self)
@property
def plan_details(self) -> "PlanDetailsEndpoint":
from pyironscales.endpoints.ironscales.PlanDetailsEndpoint import PlanDetailsEndpoint
return PlanDetailsEndpoint(self)
@property
def settings(self) -> "SettingsEndpoint":
from pyironscales.endpoints.ironscales.SettingsEndpoint import SettingsEndpoint
return SettingsEndpoint(self)
def _get_url(self) -> str:
"""
Generates and returns the URL for the Ironscales API endpoints based on the company url and codebase.

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import Campaigns
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class CampaignsIdDetailsEndpoint(
IronscalesEndpoint,
IGettable[Campaigns, IronscalesRequestParams],
IPaginateable[Campaigns, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "details", parent_endpoint=parent_endpoint)
IGettable.__init__(self, Campaigns)
IPaginateable.__init__(self, Campaigns)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[Campaigns]:
"""
Performs a GET request against the /campaigns/{id}/details endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[Campaigns]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
Campaigns,
self,
"campaigns",
page,
params,
)
def get(
self,

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import CampaignParticipants
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class CampaignsIdParticipantDetailsEndpoint(
IronscalesEndpoint,
IGettable[CampaignParticipants, IronscalesRequestParams],
IPaginateable[CampaignParticipants, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "participantdetails", parent_endpoint=parent_endpoint)
IronscalesEndpoint.__init__(self, client, "participant-details", parent_endpoint=parent_endpoint)
IGettable.__init__(self, CampaignParticipants)
IPaginateable.__init__(self, CampaignParticipants)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[CampaignParticipants]:
"""
Performs a GET request against the /campaigns/{id}/participant-details endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[CampaignParticipants]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
CampaignParticipants,
self,
"participants",
page,
params,
)
def get(
self,
@ -23,7 +54,7 @@ class CampaignsIdParticipantDetailsEndpoint(
params: IronscalesRequestParams | None = None,
) -> CampaignParticipants:
"""
Performs a GET request against the /campaigns/{id}/participantdetails endpoint.
Performs a GET request against the /campaigns/{id}/participant-details endpoint.
Parameters:
data (dict[str, Any]): The data to send in the request body.

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import CompanyMailboxes
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class CompanyIdAutoSyncMailboxesEndpoint(
IronscalesEndpoint,
IGettable[CompanyMailboxes, IronscalesRequestParams],
IPaginateable[CompanyMailboxes, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "mailboxes/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, CompanyMailboxes)
IPaginateable.__init__(self, CompanyMailboxes)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[CompanyMailboxes]:
"""
Performs a GET request against the /company/{id}/auto-sync/mailboxes/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[CompanyMailboxes]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
CompanyMailboxes,
self,
"emails",
page,
params,
)
def get(
self,

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import PartnerCompanyV2
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class CompanyListV2Endpoint(
IronscalesEndpoint,
IGettable[PartnerCompanyV2, IronscalesRequestParams],
IPaginateable[PartnerCompanyV2, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "v2", parent_endpoint=parent_endpoint)
IronscalesEndpoint.__init__(self, client, "v2/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, PartnerCompanyV2)
IPaginateable.__init__(self, PartnerCompanyV2)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[PartnerCompanyV2]:
"""
Performs a GET request against the /company/list/v2/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[PartnerCompanyV2]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
PartnerCompanyV2,
self,
"data",
page,
params,
)
def get(
self,
@ -23,7 +54,7 @@ class CompanyListV2Endpoint(
params: IronscalesRequestParams | None = None,
) -> PartnerCompanyV2:
"""
Performs a GET request against the /company/list/v2 endpoint.
Performs a GET request against the /company/list/v2/ endpoint.
Parameters:
data (dict[str, Any]): The data to send in the request body.

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import EscalatedEmails
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class EmailsIdEndpoint(
IronscalesEndpoint,
IGettable[EscalatedEmails, IronscalesRequestParams],
IPaginateable[EscalatedEmails, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "{id}", parent_endpoint=parent_endpoint)
IGettable.__init__(self, EscalatedEmails)
IPaginateable.__init__(self, EscalatedEmails)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[EscalatedEmails]:
"""
Performs a GET request against the /incident/{id}/list/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[EscalatedEmails]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
EscalatedEmails,
self,
"emails",
page,
params,
)
def get(
self,
@ -31,7 +62,7 @@ class EmailsIdEndpoint(
Returns:
EscalatedEmails: The parsed response data.
"""
return self._parse_one(
return self._parse_many(
EscalatedEmails,
super()._make_request("GET", data=data, params=params).json(),
super()._make_request("GET", data=data, params=params).json().get('emails', {}),
)

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import Incidents
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class IncidentIdListEndpoint(
IronscalesEndpoint,
IGettable[Incidents, IronscalesRequestParams],
IPaginateable[Incidents, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "list/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, Incidents)
IPaginateable.__init__(self, Incidents)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[Incidents]:
"""
Performs a GET request against the /incident/{id}/list/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[Incidents]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
Incidents,
self,
"incidents",
page,
params,
)
def get(
self,

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import ScanbackIncidents
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class IncidentIdScanbackListEndpoint(
IronscalesEndpoint,
IGettable[ScanbackIncidents, IronscalesRequestParams],
IPaginateable[ScanbackIncidents, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "scanback-list/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, ScanbackIncidents)
IPaginateable.__init__(self, ScanbackIncidents)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[ScanbackIncidents]:
"""
Performs a GET request against the /incident/{id}/scanback-list/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[ScanbackIncidents]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
ScanbackIncidents,
self,
"incidents",
page,
params,
)
def get(
self,
@ -23,7 +54,7 @@ class IncidentIdScanbackListEndpoint(
params: IronscalesRequestParams | None = None,
) -> ScanbackIncidents:
"""
Performs a GET request against the /incident/{id}/scanback-list endpoint.
Performs a GET request against the /incident/{id}/scanback-list/ endpoint.
Parameters:
data (dict[str, Any]): The data to send in the request body.

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import ComplianceReport
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class MailboxesIdComplianceReportEndpoint(
IronscalesEndpoint,
IGettable[ComplianceReport, IronscalesRequestParams],
IPaginateable[ComplianceReport, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "compliance-report/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, ComplianceReport)
IPaginateable.__init__(self, ComplianceReport)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[ComplianceReport]:
"""
Performs a GET request against the /mailboxes/{id}/compliance-report/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[ComplianceReport]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
ComplianceReport,
self,
"data",
page,
params,
)
def get(
self,
@ -23,7 +54,7 @@ class MailboxesIdComplianceReportEndpoint(
params: IronscalesRequestParams | None = None,
) -> ComplianceReport:
"""
Performs a GET request against the /incident/{id}/{status}/ endpoint.
Performs a GET request against the /mailboxes/{id}/compliance-report/ endpoint.
Parameters:
data (dict[str, Any]): The data to send in the request body.
@ -31,7 +62,7 @@ class MailboxesIdComplianceReportEndpoint(
Returns:
ComplianceReport: The parsed response data.
"""
return self._parse_one(
return self._parse_many(
ComplianceReport,
super()._make_request("GET", data=data, params=params).json(),
super()._make_request("GET", data=data, params=params).json().get('data', {}),
)

View File

@ -1,9 +1,11 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPuttable
IPuttable,
IPaginateable,
)
from pyironscales.models.ironscales import CompanyMailboxes, CompanyMailboxesPutResponse
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -14,11 +16,40 @@ class MailboxesIdListEndpoint(
IronscalesEndpoint,
IGettable[CompanyMailboxes, IronscalesRequestParams],
IPuttable[CompanyMailboxes, IronscalesRequestParams],
IPaginateable[CompanyMailboxes, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "list/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, CompanyMailboxes)
IPuttable.__init__(self, CompanyMailboxes)
IPaginateable.__init__(self, CompanyMailboxes)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[CompanyMailboxes]:
"""
Performs a GET request against the /mailboxes/{id}/list/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[CompanyMailboxes]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
CompanyMailboxes,
self,
"mailboxes",
page,
params,
)
def get(
self,
@ -36,7 +67,7 @@ class MailboxesIdListEndpoint(
"""
return self._parse_many(
CompanyMailboxes,
super()._make_request("GET", data=data, params=params).json().get('incidents', {}),
super()._make_request("GET", data=data, params=params).json().get('mailboxes', {}),
)
def put(

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import UserCampaignPerformance
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -12,10 +14,39 @@ from pyironscales.types import (
class MailboxesIdUserCampaignsPerformanceEndpoint(
IronscalesEndpoint,
IGettable[UserCampaignPerformance, IronscalesRequestParams],
IPaginateable[UserCampaignPerformance, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "user-campaigns-performance/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, UserCampaignPerformance)
IPaginateable.__init__(self, UserCampaignPerformance)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[UserCampaignPerformance]:
"""
Performs a GET request against the /mailboxes/{id}/user-campaigns-performance/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[UserCampaignPerformance]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
UserCampaignPerformance,
self,
"data",
page,
params,
)
def get(
self,
@ -33,5 +64,5 @@ class MailboxesIdUserCampaignsPerformanceEndpoint(
"""
return self._parse_many(
UserCampaignPerformance,
super()._make_request("GET", data=data, params=params).json().get('incidents', {}),
super()._make_request("GET", data=data, params=params).json().get('data', {}),
)

View File

@ -2,8 +2,10 @@ from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPostable,
IPaginateable,
)
from pyironscales.models.ironscales import CompanyImpersonationIncidents
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -13,11 +15,40 @@ class MitigationsIdImpersonationDetailsEndpoint(
IronscalesEndpoint,
IGettable[CompanyImpersonationIncidents, IronscalesRequestParams],
IPostable[CompanyImpersonationIncidents, IronscalesRequestParams],
IPaginateable[CompanyImpersonationIncidents, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "details/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, CompanyImpersonationIncidents)
IPostable.__init__(self, CompanyImpersonationIncidents)
IPaginateable.__init__(self, CompanyImpersonationIncidents)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[CompanyImpersonationIncidents]:
"""
Performs a GET request against the /mitigations/{id}/impersonation/details/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[CompanyImpersonationIncidents]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
CompanyImpersonationIncidents,
self,
"incidents",
page,
params,
)
def get(
self,

View File

@ -1,8 +1,10 @@
from pyironscales.endpoints.base.base_endpoint import IronscalesEndpoint
from pyironscales.interfaces import (
IGettable,
IPaginateable,
)
from pyironscales.models.ironscales import IncidentDetails
from pyironscales.responses.paginated_response import PaginatedResponse
from pyironscales.types import (
JSON,
IronscalesRequestParams,
@ -11,10 +13,39 @@ from pyironscales.types import (
class MitigationsIdIncidentsDetailsEndpoint(
IronscalesEndpoint,
IGettable[IncidentDetails, IronscalesRequestParams],
IPaginateable[IncidentDetails, IronscalesRequestParams],
):
def __init__(self, client, parent_endpoint=None) -> None:
IronscalesEndpoint.__init__(self, client, "details/", parent_endpoint=parent_endpoint)
IGettable.__init__(self, IncidentDetails)
IPaginateable.__init__(self, IncidentDetails)
def paginated(
self,
page: int,
params: IronscalesRequestParams | None = None,
) -> PaginatedResponse[IncidentDetails]:
"""
Performs a GET request against the /mitigations/{id}/incidents/details/ endpoint and returns an initialized PaginatedResponse object.
Parameters:
page (int): The page number to request.
params (dict[str, int | str]): The parameters to send in the request query string.
Returns:
PaginatedResponse[IncidentDetails]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
else:
params = {"page": page}
return PaginatedResponse(
super()._make_request("GET", params=params),
IncidentDetails,
self,
"incidents",
page,
params,
)
def get(
self,

View File

@ -42,7 +42,6 @@ class PaginatedResponse(Generic[TModel]):
endpointmodel: IPaginateable,
endpoint: str,
page: int,
limit: int,
params: RequestParams | None = None,
) -> None:
"""
@ -58,7 +57,7 @@ class PaginatedResponse(Generic[TModel]):
expected model type for the response data. This allows for type-safe handling
of model instances throughout the class.
"""
self._initialize(response, response_model, endpointmodel, endpoint, page, limit, params)
self._initialize(response, response_model, endpointmodel, endpoint, page, params)
def _initialize(
self,
@ -67,7 +66,6 @@ class PaginatedResponse(Generic[TModel]):
endpointmodel: IPaginateable,
endpoint: str,
page: int,
limit: int,
params: RequestParams | None = None,
):
"""
@ -77,19 +75,18 @@ class PaginatedResponse(Generic[TModel]):
response: The raw response object from the API.
endpointmodel (IronscalesEndpoint[TModel]): The endpointmodel associated with the response.
endpoint: The endpoint url to extract the data
limit (int): The number of items per page.
"""
self.response = response
self.response_model = response_model
self.endpointmodel = endpointmodel
self.endpoint = endpoint
self.limit = limit
# Get page data from the response body
try:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('pagination', {}))
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')))
except:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('meta.page', {}))
pass
self.params = params
try:
if self.parsed_pagination_response is not None:
# Ironscales API gives us a handy response to parse for Pagination
self.has_next_page: bool = self.parsed_pagination_response.get("has_next_page", False)
@ -106,6 +103,8 @@ class PaginatedResponse(Generic[TModel]):
self.prev_page = page - 1 if page > 1 else 1
self.next_page = page + 1
self.last_page = 999999
except:
pass
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
@ -122,14 +121,13 @@ class PaginatedResponse(Generic[TModel]):
self.has_data = False
return self
next_response = self.endpointmodel.paginated(self.next_page, self.limit, self.params)
next_response = self.endpointmodel.paginated(self.next_page, 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
@ -146,13 +144,12 @@ class PaginatedResponse(Generic[TModel]):
self.has_data = False
return self
prev_response = self.endpointmodel.paginated(self.prev_page, self.limit, self.params)
prev_response = self.endpointmodel.paginated(self.prev_page, self.params)
self._initialize(
prev_response.response,
prev_response.response_model,
prev_response.endpointmodel,
self.prev_page,
prev_response.limit,
self.params,
)
return self

View File

@ -52,20 +52,20 @@ def parse_response_body(
print(pagination_info)
# Output: {'first_page': 1, 'next_page': 2, 'has_next_page': True}
"""
if body.get("current_page") is None:
if body.get("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
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
total_pages: int | None = None
result = {}
@ -74,41 +74,25 @@ def parse_response_body(
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
elif body.get("page") is not None:
if body.get("page") > 1:
result["prev_page"] = body.get("page") - 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("total_pages") is not None:
result["last_page"] = body.get("total_pages")
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:
if body.get("page") is not None and body.get("page") == body.get("total_pages"):
result["has_next_page"] = False
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
elif body.get("page") is not None and body.get("page") < body.get("total_pages"):
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:
if body.get("page") is not None:
if body.get("page") > 1:
result["has_prev_page"] = True
return result