fix: exception handling in open api merger

icon_stats/api/v1/endpoints/openapi.py

@@ -1,6 +1,8 @@
 from datetime import datetime
 from typing import Dict, Optional, Any, List
+
 from fastapi import APIRouter
+
 from icon_stats.config import config
 from icon_stats.openapi.operations import FetchSchema, ResolveRefs, ValidateParams
 from icon_stats.openapi.processor import OpenAPIProcessor

icon_stats/openapi/operations.py

@@ -1,7 +1,9 @@
 import time
-from typing import Any, Dict
+from typing import Any, Dict, Optional
+
 import requests
 from pydantic import BaseModel
+
 from icon_stats.log import logger
 
 
@@ -11,7 +13,7 @@ def execute(self, *args, **kwargs) -> Any:
 
 
 class FetchSchema(OpenAPIOperation):
-    def execute(self, url: str) -> Dict[str, Any]:
+    def execute(self, url: str) -> Optional[Dict[str, Any]]:
         max_retries = 10  # Predefined maximum number of retries
         retry_delay = 2  # Delay between retries in seconds
         retries = 0
@@ -25,25 +27,25 @@ def execute(self, url: str) -> Dict[str, Any]:
                     },
                 )
 
+                # If successful, return the response data
                 if response.status_code == 200:
                     return response.json()
 
+                logger.error(f"Failed: status code: {response.status_code} for URL: {url} response : {response.json()}")
+                return None
+
             except Exception as e:
+                # Only retry on exceptions
                 retries += 1
                 if retries < max_retries:
                     logger.warning(f"Retrying {retries}/{max_retries} after error: {e}")
                     time.sleep(retry_delay)
                 else:
-                    logger.error(
-                        f"Max retries exceeded for URL: {url}. Last error: {e}"
-                    )
-                    raise Exception(
-                        f"Max retries exceeded for URL: {url}. Last error: {e}"
-                    )
-
+                    logger.error(f"Max retries exceeded for URL: {url}. Last error: {e}")
+                    return None
 
 class ResolveRefs(OpenAPIOperation):
-    def execute(self, openapi_json: Dict[str, Any], base_url: str) -> Dict[str, Any]:
+    def execute(self, openapi_json: Dict[str, Any], base_url: str) -> Optional[Dict[str, Any]]:
         def _resolve(obj, url):
             if isinstance(obj, dict):
                 if "$ref" in obj:
@@ -55,7 +57,8 @@ def _resolve(obj, url):
                         if ref_response.status_code == 200:
                             ref_obj = ref_response.json()
                         else:
-                            raise Exception(f"Reference url={ref_url} not found.")
+                            logger.error(f"Reference URL not found: {ref_url}")
+                            return None
                         return _resolve(ref_obj, url)
                     else:
                         # internal reference
@@ -65,13 +68,20 @@ def _resolve(obj, url):
                         for part in ref_parts:
                             ref_obj = ref_obj.get(part)
                             if ref_obj is None:
-                                raise KeyError(f"Reference path not found: {ref_path}")
+                                logger.error(f"Reference path not found: {ref_path}")
+                                return None
                         return _resolve(ref_obj, url)
                 else:
                     for key, value in obj.items():
-                        obj[key] = _resolve(value, url)
+                        resolved_value = _resolve(value, url)
+                        if resolved_value is None:
+                            return None
+                        obj[key] = resolved_value
             elif isinstance(obj, list):
-                return [_resolve(item, url) for item in obj]
+                resolved_list = [_resolve(item, url) for item in obj]
+                if None in resolved_list:
+                    return None
+                return resolved_list
             return obj
 
         return _resolve(openapi_json, base_url)

icon_stats/openapi/processor.py

@@ -1,9 +1,12 @@
 from typing import List, Dict, Any
 
 import requests
-from pydantic import BaseModel
 from openapi_pydantic import OpenAPI, Info, PathItem
+from pydantic import BaseModel
+
+from icon_stats.log import logger
 from .operations import FetchSchema, ResolveRefs, ValidateParams
+from ..config import config
 
 IGNORED_PATHS = [
     "/health",
@@ -26,35 +29,47 @@ def process(self, schema_urls: List[str], title: str) -> OpenAPI:
                 title=title,
                 version="v0.0.1",
             ),
+            servers=[
+                {
+                    "url": config.COMMUNITY_API_ENDPOINT
+                },
+            ],
             paths={},
         )
 
         for url in schema_urls:
-            base_url = url.rsplit('/', 1)[0]
+            base_url = url.rsplit("/", 1)[0]
             openapi_json = self.fetch_schema.execute(url)
+
+            if openapi_json is None:
+                logger.info(f"Empty schema returned for URL: {url}")
+                continue
+
             openapi_json = check_openapi_version_and_convert(schema_json=openapi_json)
             openapi_json = self.resolve_schema_refs.execute(
                 openapi_json=openapi_json, base_url=base_url
             )
             openapi_json = self.validate_params.execute(openapi_json=openapi_json)
 
-            for path_name, operations in openapi_json['paths'].items():
+            for path_name, operations in openapi_json["paths"].items():
                 if path_name in IGNORED_PATHS:
                     continue
                 if path_name in output.paths:
-                    raise Exception(
-                        f"Overlapping paths not supported (TODO) - {path_name}")
+                    logger.error(
+                        f"Overlapping paths not supported (TODO) - {path_name}"
+                    )
                 output.paths[path_name] = PathItem(**operations)
 
         return output
 
 
 def check_openapi_version_and_convert(schema_json: Dict[str, Any]) -> Dict:
-    version = schema_json.get('openapi') or schema_json.get('swagger')
+    version = schema_json.get("openapi") or schema_json.get("swagger")
     if not version:
-        raise ValueError("The schema does not have a valid OpenAPI or Swagger version.")
+        logger.error("The schema does not have a valid OpenAPI or Swagger version.")
+        return {}
 
-    major_version = int(version.split('.')[0])
+    major_version = int(version.split(".")[0])
     if major_version < 3:
         print(f"Converting OpenAPI version {version} to OpenAPI 3.x.x")