Merge pull request #7 from sudoblockio/openapi-merger

OpenAPI merger endpoint

.github/workflows/release.yml

@@ -120,9 +120,6 @@ jobs:
           - cluster: prod-ams
             network_name: mainnet
             network_version: v2
-          - cluster: prod-ams
-            network_name: sejong
-            network_version: v2
           - cluster: prod-ams
             network_name: lisbon
             network_version: v2

icon_stats/api/v1/endpoints/openapi.py

@@ -0,0 +1,57 @@
+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
+
+router = APIRouter()
+
+_cache: Dict[str, Optional[Any]] = {
+    "data": None,
+    "last_updated": None,
+    "title": "ICON"
+}
+
+
+@router.get("/openapi")
+async def get_merged_openapi_spec() -> dict:
+    """Combine the openapi specs from multiple data sources by using a cache."""
+    now = datetime.now()
+    if _cache["data"] is not None and _cache["last_updated"] is not None:
+        elapsed_time = (now - _cache["last_updated"]).total_seconds()
+        if elapsed_time < config.CACHE_DURATION:
+            return _cache["data"]
+
+    endpoints_suffixes = [
+        'api/v1/docs/doc.json',
+        'api/v1/governance/docs/openapi.json',
+        'api/v1/contracts/docs/openapi.json',
+        'api/v1/statistics/docs/openapi.json',
+    ]
+
+    schema_urls = get_openapi_urls(endpoint_suffixes=endpoints_suffixes,
+                                   base_url=config.OPENAPI_ENDPOINT_PREFIX)
+
+    output = get_merged_openapi(schema_urls=schema_urls)
+
+    # Update the cache
+    _cache["data"] = output
+    _cache["last_updated"] = now
+
+    return output
+
+
+def get_openapi_urls(endpoint_suffixes: List[str], base_url: str) -> List[str]:
+    return [f"{base_url}/{suffix}" for suffix in endpoint_suffixes]
+
+
+def get_merged_openapi(schema_urls: List[str], title: str = _cache['title']) -> Dict:
+    schema_processor = OpenAPIProcessor(
+        fetch_schema=FetchSchema(),
+        resolve_schema_refs=ResolveRefs(),
+        validate_params=ValidateParams()
+    )
+    schemas = schema_processor.process(schema_urls=schema_urls, title=title)
+
+    return schemas.model_dump(by_alias=True, exclude_none=True)

icon_stats/api/v1/router.py

@@ -1,7 +1,8 @@
 from fastapi import APIRouter
 
-from icon_stats.api.v1.endpoints import exchanges_legacy, token_stats
+from icon_stats.api.v1.endpoints import exchanges_legacy, token_stats, openapi
 
 api_router = APIRouter(prefix="/statistics")
 api_router.include_router(token_stats.router)
 api_router.include_router(exchanges_legacy.router)
+api_router.include_router(openapi.router)

icon_stats/config.py

@@ -71,6 +71,10 @@ class Settings(BaseSettings):
     LOG_INCLUDE_FIELDS: list[str] = ["timestamp", "message"]
     LOG_EXCLUDE_FIELDS: list[str] = []
 
+    # OpenAPI Merger
+    CACHE_DURATION: int = 300  # In seconds - 5 min
+    OPENAPI_ENDPOINT_PREFIX: str = "https://tracker.icon.community"
+
     model_config = SettingsConfigDict(
         case_sensitive=False,
     )

icon_stats/openapi/operations.py

@@ -0,0 +1,71 @@
+from typing import Any, Dict
+import requests
+from pydantic import BaseModel
+
+
+class OpenAPIOperation(BaseModel):
+    def execute(self, *args, **kwargs) -> Any:
+        pass
+
+
+class FetchSchema(OpenAPIOperation):
+    def execute(self, url: str) -> Dict[str, Any]:
+        response = requests.get(url=url)
+        if response.status_code == 200:
+            return response.json()
+        else:
+            raise Exception(
+                f"Failed to Fetch URL : {url} with status code {response.status_code}")
+
+
+class ResolveRefs(OpenAPIOperation):
+    def execute(self, openapi_json: Dict[str, Any], base_url: str) -> Dict[str, Any]:
+        def _resolve(obj, url):
+            if isinstance(obj, dict):
+                if '$ref' in obj:
+                    ref_path = obj['$ref']
+                    if not ref_path.startswith('#'):
+                        # external reference
+                        ref_url = f"{url}/{ref_path}"
+                        ref_response = requests.get(ref_url)
+                        if ref_response.status_code == 200:
+                            ref_obj = ref_response.json()
+                        else:
+                            raise Exception(f"Reference url={ref_url} not found.")
+                        return _resolve(ref_obj, url)
+                    else:
+                        # internal reference
+                        ref_path = ref_path.lstrip('#/')
+                        ref_parts = ref_path.split('/')
+                        ref_obj = openapi_json
+                        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}")
+                        return _resolve(ref_obj, url)
+                else:
+                    for key, value in obj.items():
+                        obj[key] = _resolve(value, url)
+            elif isinstance(obj, list):
+                return [_resolve(item, url) for item in obj]
+            return obj
+
+        return _resolve(openapi_json, base_url)
+
+
+class ValidateParams(OpenAPIOperation):
+    def execute(self, openapi_json: Dict[str, Any]) -> Dict[str, Any]:
+        def _validate(obj):
+            if isinstance(obj, dict):
+                if 'parameters' in obj:
+                    for param in obj['parameters']:
+                        if 'content' not in param and 'schema' not in param:
+                            param['schema'] = {"type": "string"}  # Default schema type
+                for key, value in obj.items():
+                    _validate(value)
+            elif isinstance(obj, list):
+                for item in obj:
+                    _validate(item)
+
+        _validate(openapi_json)
+        return openapi_json

icon_stats/openapi/processor.py

@@ -0,0 +1,66 @@
+from typing import List, Dict, Any
+
+import requests
+from pydantic import BaseModel
+from openapi_pydantic import OpenAPI, Info, PathItem
+from .operations import FetchSchema, ResolveRefs, ValidateParams
+
+IGNORED_PATHS = [
+    "/health",
+    "/ready",
+    "/metadata",
+    "/version",
+]
+
+SWAGGER_CONVERT = "https://converter.swagger.io/api/convert"
+
+
+class OpenAPIProcessor(BaseModel):
+    fetch_schema: FetchSchema
+    resolve_schema_refs: ResolveRefs
+    validate_params: ValidateParams
+
+    def process(self, schema_urls: List[str], title: str) -> OpenAPI:
+        output = OpenAPI(
+            info=Info(
+                title=title,
+                version="v0.0.1",
+            ),
+            paths={},
+        )
+
+        for url in schema_urls:
+            base_url = url.rsplit('/', 1)[0]
+            openapi_json = self.fetch_schema.execute(url)
+            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():
+                if path_name in IGNORED_PATHS:
+                    continue
+                if path_name in output.paths:
+                    raise Exception(
+                        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')
+    if not version:
+        raise ValueError("The schema does not have a valid OpenAPI or Swagger version.")
+
+    major_version = int(version.split('.')[0])
+    if major_version < 3:
+        print(f"Converting OpenAPI version {version} to OpenAPI 3.x.x")
+
+        response = requests.post(url=SWAGGER_CONVERT, json=schema_json)
+        if response.status_code == 200:
+            return response.json()
+
+    else:
+        return schema_json

requirements-api.txt

@@ -4,4 +4,6 @@ uvicorn==0.23.2
 fastapi_health
 brotli-asgi~=1.1.0
 #starlette~=0.14.2
-#starlette
+#starlette
+
+openapi_pydantic

tests/conftest.py

@@ -79,13 +79,14 @@ def get_env(self, filepath):
     def _set_env_from_file(self, filepath) -> list[tuple[str, str]]:
         """Get list of tuples from an env file to temporarily set them on tests."""
         env_vars = []
-        with open(filepath, "r") as f:
-            for line in f:
-                line = line.strip()  # Remove leading and trailing whitespace
-                if line and not line.startswith("#"):  # Ignore empty lines and comments
-                    key, value = line.split("=", 1)
-                    value = value.strip().strip('"').strip("'")
-                    env_vars.append((key, value))
+        if os.path.isfile(filepath):
+            with open(filepath, "r") as f:
+                for line in f:
+                    line = line.strip()  # Remove leading and trailing whitespace
+                    if line and not line.startswith("#"):  # Ignore empty lines and comments
+                        key, value = line.split("=", 1)
+                        value = value.strip().strip('"').strip("'")
+                        env_vars.append((key, value))
         return env_vars
 
 

tests/integration/api/test_api_markets.py

@@ -4,7 +4,7 @@
 
 
 def test_api_get_markets(client: TestClient):
-    response = client.get(f"{config.API_REST_PREFIX}/stats/exchanges/legacy")
+    response = client.get(f"{config.API_REST_PREFIX}/statistics/exchanges/legacy")
     assert response.status_code == 200
     assert response.json()['data']['marketCap'] > 10000000