PORT-4337 Improved error PortClient handling

changelog/PORT-4337.improvement.md

@@ -0,0 +1 @@
+Improved error messages from the PortClient

port_ocean/clients/port/authentication.py

@@ -5,6 +5,7 @@
 from pydantic import BaseModel, Field, PrivateAttr
 
 from port_ocean.clients.port.types import UserAgentType
+from port_ocean.clients.port.utils import handle_status_code
 from port_ocean.utils import get_time
 
 
@@ -45,11 +46,11 @@ async def _get_token(self, client_id: str, client_secret: str) -> TokenResponse:
         logger.info(f"Fetching access token for clientId: {client_id}")
 
         credentials = {"clientId": client_id, "clientSecret": client_secret}
-        token_response = await self.client.post(
+        response = await self.client.post(
             f"{self.api_url}/auth/access_token", json=credentials
         )
-        token_response.raise_for_status()
-        return TokenResponse(**token_response.json())
+        handle_status_code(response)
+        return TokenResponse(**response.json())
 
     def user_agent(self, user_agent_type: UserAgentType | None = None) -> str:
         user_agent = f"port-ocean/{self.integration_type}/{self.integration_identifier}"

port_ocean/clients/port/client.py

@@ -37,14 +37,14 @@ def __init__(
         )
         BlueprintClientMixin.__init__(self, self.auth, self.client)
 
-    async def get_kafka_creds(self, silent: bool = False) -> KafkaCreds:
+    async def get_kafka_creds(self) -> KafkaCreds:
         logger.info("Fetching organization kafka credentials")
         response = await self.client.get(
             f"{self.api_url}/kafka-credentials", headers=await self.auth.headers()
         )
         if response.is_error:
-            logger.error(f"Error getting kafka credentials, error: {response.text}")
-        handle_status_code(silent, response)
+            logger.error("Error getting kafka credentials")
+        handle_status_code(response)
 
         credentials = response.json().get("credentials")
 
@@ -61,6 +61,6 @@ async def get_org_id(self) -> str:
         )
         if response.is_error:
             logger.error(f"Error getting organization id, error: {response.text}")
-            response.raise_for_status()
+        handle_status_code(response)
 
         return response.json()["organization"]["id"]

port_ocean/clients/port/mixins/blueprints.py

@@ -13,29 +13,27 @@ def __init__(self, auth: PortAuthentication, client: httpx.AsyncClient):
         self.auth = auth
         self.client = client
 
-    async def get_blueprint(self, identifier: str, silent: bool = False) -> Blueprint:
+    async def get_blueprint(self, identifier: str) -> Blueprint:
         logger.info(f"Fetching blueprint with id: {identifier}")
         response = await self.client.get(
             f"{self.auth.api_url}/blueprints/{identifier}",
             headers=await self.auth.headers(),
         )
-        handle_status_code(silent, response)
+        handle_status_code(response)
         return Blueprint.parse_obj(response.json()["blueprint"])
 
-    async def create_blueprint(
-        self, raw_blueprint: dict[str, Any], silent: bool = False
-    ) -> None:
+    async def create_blueprint(self, raw_blueprint: dict[str, Any]) -> None:
         logger.info(f"Creating blueprint with id: {raw_blueprint.get('identifier')}")
         headers = await self.auth.headers()
         response = await self.client.post(
             f"{self.auth.api_url}/blueprints", headers=headers, json=raw_blueprint
         )
-        handle_status_code(silent, response)
+        handle_status_code(response)
         if response.is_success:
             return response.json()["blueprint"]
 
     async def patch_blueprint(
-        self, identifier: str, raw_blueprint: dict[str, Any], silent: bool = False
+        self, identifier: str, raw_blueprint: dict[str, Any]
     ) -> None:
         logger.info(f"Patching blueprint with id: {identifier}")
         headers = await self.auth.headers()
@@ -44,19 +42,21 @@ async def patch_blueprint(
             headers=headers,
             json=raw_blueprint,
         )
-        handle_status_code(silent, response)
+        handle_status_code(response)
 
-    async def delete_blueprint(self, identifier: str, silent: bool = False) -> None:
+    async def delete_blueprint(
+        self, identifier: str, should_raise: bool = False
+    ) -> None:
         logger.info(f"Deleting blueprint with id: {identifier}")
         headers = await self.auth.headers()
         response = await self.client.delete(
             f"{self.auth.api_url}/blueprints/{identifier}",
             headers=headers,
         )
-        handle_status_code(silent, response)
+        handle_status_code(response, should_raise)
 
     async def create_action(
-        self, blueprint_identifier: str, action: dict[str, Any], silent: bool = False
+        self, blueprint_identifier: str, action: dict[str, Any]
     ) -> None:
         logger.info(f"Creating action: {action}")
         response = await self.client.post(
@@ -65,10 +65,12 @@ async def create_action(
             headers=await self.auth.headers(),
         )
 
-        handle_status_code(silent, response)
+        handle_status_code(response)
 
     async def create_scorecard(
-        self, blueprint_identifier: str, scorecard: dict[str, Any], silent: bool = False
+        self,
+        blueprint_identifier: str,
+        scorecard: dict[str, Any],
     ) -> None:
         logger.info(f"Creating scorecard: {scorecard}")
         response = await self.client.post(
@@ -77,4 +79,4 @@ async def create_scorecard(
             headers=await self.auth.headers(),
         )
 
-        handle_status_code(silent, response)
+        handle_status_code(response)

port_ocean/clients/port/mixins/entities.py

@@ -19,7 +19,7 @@ async def upsert_entity(
         entity: Entity,
         request_options: RequestOptions,
         user_agent_type: UserAgentType | None = None,
-        silent: bool = False,
+        should_raise: bool = True,
     ) -> None:
         validation_only = request_options.get("validation_only", False)
         logger.info(
@@ -45,17 +45,16 @@ async def upsert_entity(
             logger.error(
                 f"Error {'Validating' if validation_only else 'Upserting'} "
                 f"entity: {entity.identifier} of "
-                f"blueprint: {entity.blueprint}, "
-                f"error: {response.text}"
+                f"blueprint: {entity.blueprint}"
             )
-        handle_status_code(silent, response)
+        handle_status_code(response, should_raise)
 
     async def delete_entity(
         self,
         entity: Entity,
         request_options: RequestOptions,
         user_agent_type: UserAgentType | None = None,
-        silent: bool = False,
+        should_raise: bool = True,
     ) -> None:
         logger.info(
             f"Delete entity: {entity.identifier} of blueprint: {entity.blueprint}"
@@ -74,11 +73,10 @@ async def delete_entity(
             logger.error(
                 f"Error deleting "
                 f"entity: {entity.identifier} of "
-                f"blueprint: {entity.blueprint}, "
-                f"error: {response.text}"
+                f"blueprint: {entity.blueprint}"
             )
 
-        handle_status_code(silent, response)
+        handle_status_code(response, should_raise)
 
     async def validate_entity_exist(self, identifier: str, blueprint: str) -> None:
         logger.info(f"Validating entity {identifier} of blueprint {blueprint} exists")
@@ -91,10 +89,9 @@ async def validate_entity_exist(self, identifier: str, blueprint: str) -> None:
             logger.error(
                 f"Error validating "
                 f"entity: {identifier} of "
-                f"blueprint: {blueprint}, "
-                f"error: {response.text}"
+                f"blueprint: {blueprint}"
             )
-        response.raise_for_status()
+        handle_status_code(response)
 
     async def search_entities(self, user_agent_type: UserAgentType) -> list[Entity]:
         query = {
@@ -118,7 +115,7 @@ async def search_entities(self, user_agent_type: UserAgentType) -> list[Entity]:
                 "include": ["blueprint", "identifier"],
             },
         )
-        response.raise_for_status()
+        handle_status_code(response)
         return [Entity.parse_obj(result) for result in response.json()["entities"]]
 
     async def search_dependent_entities(self, entity: Entity) -> list[Entity]:
@@ -140,7 +137,7 @@ async def search_dependent_entities(self, entity: Entity) -> list[Entity]:
             headers=await self.auth.headers(),
             json=body,
         )
-        response.raise_for_status()
+        handle_status_code(response)
 
         return [Entity.parse_obj(result) for result in response.json()["entities"]]
 

port_ocean/clients/port/mixins/integrations.py

@@ -5,6 +5,7 @@
 from starlette import status
 
 from port_ocean.clients.port.authentication import PortAuthentication
+from port_ocean.clients.port.utils import handle_status_code
 
 if TYPE_CHECKING:
     from port_ocean.core.handlers.port_app_config.models import PortAppConfig
@@ -21,13 +22,17 @@ def __init__(
         self.auth = auth
         self.client = client
 
-    async def get_current_integration(self) -> dict[str, Any]:
+    async def _get_current_integration(self) -> httpx.Response:
         logger.info(f"Fetching integration with id: {self.integration_identifier}")
         response = await self.client.get(
             f"{self.auth.api_url}/integration/{self.integration_identifier}",
             headers=await self.auth.headers(),
         )
-        response.raise_for_status()
+        return response
+
+    async def get_current_integration(self) -> dict[str, Any]:
+        response = await self._get_current_integration()
+        handle_status_code(response)
         return response.json()["integration"]
 
     async def create_integration(
@@ -48,7 +53,7 @@ async def create_integration(
         response = await self.client.post(
             f"{self.auth.api_url}/integration", headers=headers, json=json
         )
-        response.raise_for_status()
+        handle_status_code(response)
 
     async def patch_integration(
         self,
@@ -71,7 +76,7 @@ async def patch_integration(
             headers=headers,
             json=json,
         )
-        response.raise_for_status()
+        handle_status_code(response)
 
     async def initialize_integration(
         self,
@@ -80,9 +85,13 @@ async def initialize_integration(
         port_app_config: Optional["PortAppConfig"] = None,
     ) -> None:
         logger.info(f"Initiating integration with id: {self.integration_identifier}")
-        try:
-            integration = await self.get_current_integration()
+        response = await self._get_current_integration()
+        if response.status_code == status.HTTP_404_NOT_FOUND:
+            await self.create_integration(_type, changelog_destination, port_app_config)
+        else:
+            handle_status_code(response)
 
+            integration = response.json()["integration"]
             logger.info("Checking for diff in integration configuration")
             if (
                 integration["changelogDestination"] != changelog_destination
@@ -91,17 +100,6 @@ async def initialize_integration(
                 await self.patch_integration(
                     _type, changelog_destination, port_app_config
                 )
-        except httpx.HTTPStatusError as e:
-            if e.response.status_code == status.HTTP_404_NOT_FOUND:
-                await self.create_integration(
-                    _type, changelog_destination, port_app_config
-                )
-                return
-
-            logger.error(
-                f"Error initiating integration with id: {self.integration_identifier}, error: {e.response.text}"
-            )
-            raise
 
         logger.info(
             f"Integration with id: {self.integration_identifier} successfully registered"

port_ocean/clients/port/utils.py

@@ -1,6 +1,13 @@
 import httpx
+from loguru import logger
 
 
-def handle_status_code(silent: bool, response: httpx.Response) -> None:
-    if not silent:
+def handle_status_code(
+    response: httpx.Response, should_raise: bool = True, should_log: bool = True
+) -> None:
+    if should_log and response.is_error:
+        logger.error(
+            f"Request failed with status code: {response.status_code}, Error: {response.text}"
+        )
+    if should_raise:
         response.raise_for_status()

port_ocean/core/handlers/entities_state_applier/port/applier.py

@@ -159,7 +159,7 @@ async def upsert(
                         entity,
                         event.port_app_config.get_port_request_options(),
                         user_agent_type,
-                        silent=True,
+                        should_raise=False,
                     )
                     for entity in entities
                 ),
@@ -175,7 +175,7 @@ async def upsert(
                     entity,
                     event.port_app_config.get_port_request_options(),
                     user_agent_type,
-                    silent=True,
+                    should_raise=False,
                 )
 
     async def delete(
@@ -189,7 +189,7 @@ async def delete(
                         entity,
                         event.port_app_config.get_port_request_options(),
                         user_agent_type,
-                        silent=True,
+                        should_raise=False,
                     )
                     for entity in entities
                 )
@@ -202,5 +202,5 @@ async def delete(
                     entity,
                     event.port_app_config.get_port_request_options(),
                     user_agent_type,
-                    silent=True,
+                    should_raise=False,
                 )

port_ocean/port_defaults.py

@@ -1,11 +1,11 @@
 import asyncio
 import json
+from pathlib import Path
 from typing import Type, Any, TypedDict, Optional
 
 import httpx
 import yaml
 from loguru import logger
-from pathlib import Path
 from pydantic import BaseModel, Field
 from starlette import status
 
@@ -172,7 +172,9 @@ async def _initialize_defaults(
         )
         await asyncio.gather(
             *(
-                port_client.delete_blueprint(blueprint["identifier"], silent=True)
+                port_client.delete_blueprint(
+                    blueprint["identifier"], should_raise=False
+                )
                 for blueprint in defaults.blueprints
             )
         )