Paginate Pagerduty oncall (#241)

The previous implementation of the who is oncall did not use pagination. This PR improves on it. It also updates the blueprints to hold an array of users on calll

integrations/pagerduty/.port/resources/blueprints.json

@@ -17,8 +17,11 @@
             },
             "oncall": {
               "title": "On Call",
-              "type": "string",
-              "format": "user"
+              "type": "array",
+              "items": {
+                "type": "string",
+                "format": "user"
+              }
             }
           },
           "required": []

integrations/pagerduty/.port/resources/port-app-config.yaml

@@ -11,7 +11,7 @@ resources:
           properties:
             status: .status
             url: .html_url
-            oncall: .__oncall_user[0].user.email
+            oncall: '[.__oncall_user[].user.email]'
   - kind: incidents
     selector:
       query: 'true'

integrations/pagerduty/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- towncrier release notes start -->
 
+# Port_Ocean 0.1.16 (2023-11-23)
+
+### Bug Fixes
+
+- Fixed incomplete oncall list over the `Service` kind by adding pagination support to the request
+
+
 # Port_Ocean 0.1.15 (2023-11-21)
 
 ### Improvements

integrations/pagerduty/clients/pagerduty.py

@@ -142,25 +142,19 @@ async def create_webhooks_if_not_exists(self) -> None:
     async def get_oncall_user(
         self, *escalation_policy_ids: str
     ) -> list[dict[str, Any]]:
-        url = f"{self.api_url}/oncalls"
+        params = {
+            "escalation_policy_ids[]": ",".join(escalation_policy_ids),
+            "include[]": "users",
+        }
+        oncalls = []
 
-        async with httpx.AsyncClient() as client:
-            try:
-                response = await client.get(
-                    url,
-                    params={
-                        "escalation_policy_ids[]": ",".join(escalation_policy_ids),
-                        "include[]": "users",
-                    },
-                    headers=self.api_auth_header,
-                )
-                response.raise_for_status()
-                return response.json()["oncalls"]
-            except httpx.HTTPStatusError as e:
-                logger.error(
-                    f"HTTP error with status code: {e.response.status_code} and response text: {e.response.text}"
-                )
-                raise
+        async for oncall_batch in self.paginate_request_to_pager_duty(
+            data_key="oncalls", params=params
+        ):
+            logger.info(f"Received oncalls with batch size {len(oncall_batch)}")
+            oncalls.extend(oncall_batch)
+
+        return oncalls
 
     async def update_oncall_users(
         self, services: list[dict[str, Any]]

integrations/pagerduty/main.py

@@ -11,14 +11,18 @@
 from port_ocean.core.ocean_types import ASYNC_GENERATOR_RESYNC_TYPE
 
 
-@ocean.on_resync(ObjectKind.INCIDENTS)
-async def on_incidents_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
-    logger.info(f"Listing Pagerduty resource: {kind}")
-    pager_duty_client = PagerDutyClient(
+def initialize_client() -> PagerDutyClient:
+    return PagerDutyClient(
         ocean.integration_config["token"],
         ocean.integration_config["api_url"],
         ocean.integration_config.get("app_host"),
     )
+
+
+@ocean.on_resync(ObjectKind.INCIDENTS)
+async def on_incidents_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
+    logger.info(f"Listing Pagerduty resource: {kind}")
+    pager_duty_client = initialize_client()
     query_params = typing.cast(
         PagerdutyIncidentResourceConfig, event.resource_config
     ).selector.api_query_params
@@ -33,11 +37,7 @@ async def on_incidents_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
 @ocean.on_resync(ObjectKind.SERVICES)
 async def on_services_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
     logger.info(f"Listing Pagerduty resource: {kind}")
-    pager_duty_client = PagerDutyClient(
-        ocean.integration_config["token"],
-        ocean.integration_config["api_url"],
-        ocean.integration_config.get("app_host"),
-    )
+    pager_duty_client = initialize_client()
     query_params = typing.cast(
         PagerdutyServiceResourceConfig, event.resource_config
     ).selector.api_query_params
@@ -51,11 +51,7 @@ async def on_services_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
 
 @ocean.router.post("/webhook")
 async def upsert_incident_webhook_handler(data: dict[str, Any]) -> None:
-    pager_duty_client = PagerDutyClient(
-        ocean.integration_config["token"],
-        ocean.integration_config["api_url"],
-        ocean.integration_config.get("app_host"),
-    )
+    pager_duty_client = initialize_client()
     event_type = data["event"]["event_type"]
     logger.info(f"Processing Pagerduty webhook for event type: {event_type}")
     if event_type in pager_duty_client.service_delete_events:
@@ -84,10 +80,6 @@ async def on_start() -> None:
         logger.info("Skipping webhook creation because the event listener is ONCE")
         return
 
-    pager_duty_client = PagerDutyClient(
-        ocean.integration_config["token"],
-        ocean.integration_config["api_url"],
-        ocean.integration_config.get("app_host"),
-    )
+    pager_duty_client = initialize_client()
     logger.info("Subscribing to Pagerduty webhooks")
     await pager_duty_client.create_webhooks_if_not_exists()

integrations/pagerduty/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pagerduty"
-version = "0.1.15"
+version = "0.1.16"
 description = "Pagerduty Integration"
 authors = ["Port Team <[email protected]>"]