diff --git a/gallagher/cc/alarms/__init__.py b/gallagher/cc/alarms/__init__.py index 38892cc..eb6d7eb 100644 --- a/gallagher/cc/alarms/__init__.py +++ b/gallagher/cc/alarms/__init__.py @@ -15,7 +15,10 @@ from ...dto.ref import AlarmRef from ...dto.summary import AlarmSummary from ...dto.detail import AlarmDetail -from ...dto.response import AlarmSummaryResponse +from ...dto.response import ( + AlarmSummaryResponse, + AlarmUpdateResponse, +) from ...dto.payload import AlarmCommentPayload class Alarms( @@ -48,6 +51,7 @@ async def get_config(cls) -> EndpointConfig: return EndpointConfig( endpoint=Capabilities.CURRENT.features.alarms.alarms, endpoint_follow=Capabilities.CURRENT.features.alarms.updates, + dto_follow=AlarmUpdateResponse, dto_list=AlarmSummaryResponse, dto_retrieve=AlarmDetail, ) diff --git a/gallagher/cc/core.py b/gallagher/cc/core.py index e461bfd..9229623 100644 --- a/gallagher/cc/core.py +++ b/gallagher/cc/core.py @@ -150,6 +150,8 @@ class EndpointConfig: endpoint: str # partial path to the endpoint e.g. day_category endpoint_follow: str | None = None # partial path to the follow endpoint + + dto_follow: Optional[any] = None # DTO to be used for follow requests dto_list: Optional[any] = None # DTO to be used for list requests dto_retrieve: Optional[any] = None # DTO to be used for retrieve requests @@ -459,40 +461,8 @@ async def previous(cls, response): ) @classmethod - async def updates( + async def follow( cls, - params: dict[str, Any] = {}, - ): - """ Follow for updates - - - You should get no more than 100 items at a time - - If you do then you have fallen behind - - On an interval you will always get a response, even if - there are no updates - - Simply follow the next href to get the next set of updates - - See also HATEOAS on how to follow hrefs - """ - await cls._discover() - - # throw an exception if the class configuration does not - # implement an updates method - if not cls.__config__.endpoint_follow: - raise PathFollowNotSupportedError( - "Endpoint does not support updates" - ) - - return cls._follow( - cls.__config__.endpoint_follow, - cls.__config__.dto_list, - params=params, - ) - - @classmethod - async def _follow( - cls, - url: str, - response_class: AppBaseResponseWithFollowModel, event: Event, params: dict[str, Any] = {}, ): @@ -508,6 +478,16 @@ async def _follow( This behaviour is followed by updates and changes endpoints, this method should be used a helper for the updates and changes methods. """ + await cls._discover() + + if not cls.__config__.endpoint_follow: + raise PathFollowNotSupportedError( + "Endpoint does not support previous, next or updates" + ) + + # Initial url is set to endpoint_follow + url = f"{cls.__config__.endpoint_follow.href}" + async with httpx.AsyncClient(proxy=proxy_address) as _httpx_async: while event.is_set(): try: @@ -520,7 +500,7 @@ async def _follow( if response.status_code == HTTPStatus.OK: - parsed_obj = response_class.model_validate( + parsed_obj = cls.__config__.dto_follow.model_validate( response.json() )