Initial commit of openapi-analyzer, a tool for getting some info and …

…stats about the endpoints defined in an OpenAPI spec.

src/scripts/.gitignore

@@ -0,0 +1,4 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class

src/scripts/openapi-analyzer/DetailedStatsGenerator.py

@@ -0,0 +1,115 @@
+import os
+from dataclasses import dataclass, field
+from EndpointInfo import EndpointInfo
+from constants import ENDPOINT_OPERATIONS
+from openapi_pydantic.v3.v3_0 import OpenAPI, Reference
+
+
+@dataclass 
+class RequestStats:
+    num_operations: int = 0
+    num_with_body: int = 0
+    num_with_examples: int = 0
+    num_examples: int = 0
+    content_types: set[str] = field(default_factory=set)
+
+@dataclass
+class ResponseCodeStats:
+    num_operations: int = 0
+    num_with_body: int = 0
+    num_with_examples: int = 0
+    num_examples: int = 0
+    content_types: set[str] = field(default_factory=set)
+
+@dataclass 
+class ResponseStats:
+    num_operations: int = 0
+
+@dataclass
+class OperationStats:
+    request_stats: RequestStats = field(default_factory=RequestStats)
+    response_stats: dict[str, ResponseCodeStats] = field(default_factory=dict)
+
+@dataclass
+class DetailedStats:
+    num_endpoints: int = 0
+    num_operations: int = 0
+    operation_stats: dict[str, OperationStats] = field(default_factory=dict)
+
+
+class DetailedStatsGenerator:
+    def __init__(self, openapi_spec: OpenAPI):
+        self.openapi_spec = openapi_spec
+
+    def get_endpoint_info_list(self) -> list[EndpointInfo]:
+        endpoint_info_list = []
+        for path, path_item in self.openapi_spec.paths.items():
+            endpointInfo = EndpointInfo.from_path(path, path_item)
+            endpoint_info_list.append(endpointInfo)
+        return endpoint_info_list
+
+    def get_detailed_stats(self) -> DetailedStats:
+        endpoint_info_list = self.get_endpoint_info_list()
+        stats = DetailedStats()
+        stats.num_endpoints = len(endpoint_info_list)
+        for operation in ENDPOINT_OPERATIONS:
+            for endpoint_info in endpoint_info_list:
+                if operation in endpoint_info.operations:
+                    stats.num_operations += 1
+                    if operation not in stats.operation_stats:
+                        stats.operation_stats[operation] = OperationStats()
+                    operation_stats = stats.operation_stats[operation]
+                    operation_stats.request_stats.num_operations += 1
+                    if endpoint_info.operations[operation].requestBody:
+                        requestBody = endpoint_info.operations[operation].requestBody
+                        if isinstance(requestBody, Reference):
+                            component_request_ref = os.path.basename(requestBody.ref)
+                            requestBody = self.openapi_spec.components.requestBodies[component_request_ref] 
+                        for content_type, media_type in requestBody.content.items():
+                            operation_stats.request_stats.content_types.add(content_type)
+                            if media_type.examples:
+                                operation_stats.request_stats.num_with_examples += 1
+                                operation_stats.request_stats.num_examples += len(media_type.examples)
+                        operation_stats.request_stats.num_with_body += 1
+                    if endpoint_info.operations[operation].responses:
+                        for response_code, response in endpoint_info.operations[operation].responses.items():
+                            if response_code not in operation_stats.response_stats: 
+                                operation_stats.response_stats[response_code] = ResponseCodeStats()
+                            operation_stats.response_stats[response_code].num_operations += 1
+                            if isinstance(response, Reference):
+                                component_response_ref = os.path.basename(response.ref)
+                                response = self.openapi_spec.components.responses[component_response_ref]
+                            if response.content:
+                                for content_type, media_type in response.content.items():
+                                    operation_stats.response_stats[response_code].content_types.add(content_type)
+                                    if media_type.examples:
+                                        operation_stats.response_stats[response_code].num_with_examples += 1
+                                        operation_stats.response_stats[response_code].num_examples += len(media_type.examples)
+                                operation_stats.response_stats[response_code].num_with_body += 1    
+        return stats
+
+    def print_detailed_stats(self, stats: DetailedStats):
+        print("========================")
+        print("==== Detailed Stats ====")
+        print("========================")
+        print(f"Number of endpoints: {stats.num_endpoints}")
+        print(f"Number of operations: {stats.num_operations}")  
+        for operation, operation_stats in stats.operation_stats.items():
+            print(f"    {operation}: {operation_stats.request_stats.num_operations}")
+            print("        Requests:")
+            print(f"            with body: {operation_stats.request_stats.num_with_body}")
+            print("                  Content types:")
+            for content_type in operation_stats.request_stats.content_types:
+                print(f"                      {content_type}")
+            print(f"            Has examples: {operation_stats.request_stats.num_with_examples}")
+            print(f"            Number of examples: {operation_stats.request_stats.num_examples}")
+            for response_code, response_code_stats in operation_stats.response_stats.items():
+                print("        Responses:")
+                print(f"           '{response_code}': {response_code_stats.num_operations}")
+                print(f"              Has content: {response_code_stats.num_with_body}")
+                print("                  Content types:")
+                for content_type in response_code_stats.content_types:
+                    print(f"                      {content_type}")
+                print(f"              Has examples: {response_code_stats.num_with_examples}")
+                print(f"              Number of examples: {response_code_stats.num_examples}")
+        print()

src/scripts/openapi-analyzer/EndpointInfo.py

@@ -0,0 +1,24 @@
+from openapi_pydantic.v3.v3_0 import PathItem
+from constants import ENDPOINT_OPERATIONS
+
+class EndpointInfo:
+    def __init__(self, path: str):
+        self.path = path
+        self.operations = {}
+        self.summary = None
+        self.description = None
+        self.parameters = []
+
+    def init(self, path_item: PathItem):
+        self.summary = path_item.summary
+        self.description = path_item.description
+        self.parameters = path_item.parameters
+        for operation in ENDPOINT_OPERATIONS:
+            if getattr(path_item, operation):
+                self.operations[operation] = getattr(path_item, operation)
+
+    @staticmethod
+    def from_path(path: str, path_item: PathItem):
+        endpoint_info = EndpointInfo(path)
+        endpoint_info.init(path_item)
+        return endpoint_info

src/scripts/openapi-analyzer/OpenapiAnalyzer.py

@@ -0,0 +1,20 @@
+from DetailedStatsGenerator import DetailedStatsGenerator
+from SummaryStatsGenerator import SummaryStatsGenerator
+from openapi_pydantic.v3.v3_0 import OpenAPI
+
+
+class OpenapiAnalyzer:
+    def __init__(self, openapi_filepath):
+        self.openapi_filepath = openapi_filepath
+
+    def run(self):
+        openapi_spec = OpenAPI.parse_file(self.openapi_filepath)
+        print(f"OpenAPI version: {openapi_spec.openapi}\n")
+        detailed_stats_generator = DetailedStatsGenerator(openapi_spec)  
+        detailed_stats = detailed_stats_generator.get_detailed_stats()
+        summary_stats_generator = SummaryStatsGenerator(detailed_stats)
+        summary_stats = summary_stats_generator.get_summary_stats()
+        summary_stats_generator.print_summary_stats(summary_stats)
+        detailed_stats_generator.print_detailed_stats(detailed_stats)
+
+

src/scripts/openapi-analyzer/SummaryStatsGenerator.py

@@ -0,0 +1,60 @@
+from dataclasses import dataclass, field
+from DetailedStatsGenerator import DetailedStats
+
+
+@dataclass
+class SummaryStats:
+    num_endpoints: int = 0
+    num_operations: int = 0
+    num_requests_with_body: int = 0
+    num_requests_with_examples: int = 0
+    num_request_examples: int = 0
+    num_responses_with_body: int = 0
+    num_responses_with_examples: int = 0
+    num_response_examples: int = 0
+    num_examples: int = 0
+    request_content_types: set[str] = field(default_factory=set)
+    response_content_types: set[str] = field(default_factory=set)
+
+
+class SummaryStatsGenerator:
+    def __init__(self, detailed_stats: DetailedStats):
+        self.detailed_stats = detailed_stats
+
+    def get_summary_stats(self) -> SummaryStats:
+        summary_stats = SummaryStats()
+        summary_stats.num_endpoints = self.detailed_stats.num_endpoints
+        summary_stats.num_operations = self.detailed_stats.num_operations
+        for _operation, operation_stats in self.detailed_stats.operation_stats.items():
+            summary_stats.num_requests_with_body += operation_stats.request_stats.num_with_body
+            summary_stats.num_requests_with_examples += operation_stats.request_stats.num_with_examples
+            summary_stats.num_request_examples += operation_stats.request_stats.num_examples
+            summary_stats.request_content_types.update(operation_stats.request_stats.content_types)
+            for _response_code, response_code_stats in operation_stats.response_stats.items():
+                summary_stats.num_responses_with_body += response_code_stats.num_with_body
+                summary_stats.num_responses_with_examples += response_code_stats.num_with_examples
+                summary_stats.num_response_examples += response_code_stats.num_examples
+                summary_stats.response_content_types.update(response_code_stats.content_types)
+        return summary_stats
+
+    def print_summary_stats(self, stats: SummaryStats):
+        print("=======================")
+        print("==== Summary Stats ====")
+        print("=======================")
+        print(f"Number of endpoints: {stats.num_endpoints}")
+        print(f"Number of operations: {stats.num_operations}")  
+        print("    Requests:")
+        print(f"        with body: {stats.num_requests_with_body}")
+        print(f"        with examples: {stats.num_requests_with_examples}")
+        print(f"        number of examples: {stats.num_request_examples}")  
+        print("        Content types:")
+        for content_type in stats.request_content_types:
+            print(f"            {content_type}")
+        print("    Responses:")
+        print(f"        with body: {stats.num_responses_with_body}")
+        print(f'        with examples: {stats.num_responses_with_examples}')
+        print(f"        number of examples: {stats.num_response_examples}")
+        print("        Content types:")
+        for content_type in stats.response_content_types:
+            print(f"            {content_type}")    
+        print()

src/scripts/openapi-analyzer/constants.py

@@ -0,0 +1,14 @@
+DEFAULT_OUTPUT_PATH = "../../../output"
+DEFAULT_OPENAPI_FOLDER = "openapi"
+DEFAULT_OPENAPI_FILE = "elasticsearch-openapi.json"
+
+ENDPOINT_OPERATIONS = [
+    "get",
+    "put",
+    "post",
+    "delete",
+    "options",
+    "head",
+    "patch",
+    "trace"
+]

src/scripts/openapi-analyzer/main.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+
+import os
+from constants import (
+    DEFAULT_OUTPUT_PATH, 
+    DEFAULT_OPENAPI_FOLDER, 
+    DEFAULT_OPENAPI_FILE
+)
+from OpenapiAnalyzer import OpenapiAnalyzer
+
+def main():
+    openpi_filepath = os.path.join(DEFAULT_OUTPUT_PATH, 
+                                   DEFAULT_OPENAPI_FOLDER,
+                                   DEFAULT_OPENAPI_FILE)
+    if not os.path.exists(openpi_filepath):
+        print(f"OpenAPI file not found: {openpi_filepath}")
+        return
+    openapi_analyzer = OpenapiAnalyzer(openpi_filepath)
+    openapi_analyzer.run()
+
+if __name__ == "__main__":
+    main()

src/scripts/openapi-analyzer/requirements.txt

@@ -0,0 +1,2 @@
+
+openapi-pydantic==0.5.1