openapi_index.py

"""OpenAPI index for fast searching and querying."""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

import yaml


class OpenAPIIndex:
    """
    Index an OpenAPI spec for fast querying.

    Usage:
        idx = OpenAPIIndex("openapi.yaml")

        # Find endpoints
        idx.find_endpoints("Contact")
        idx.find_endpoints("Invoice.*send")

        # Get endpoint details
        idx.get_endpoint("/Invoice/{invoiceId}/sendViaEmail")

        # Get schema
        idx.get_schema("Invoice")

        # Search everything
        idx.search("statistics")
        idx.search("sum")
    """

    def __init__(self, path: str | Path) -> None:
        with open(path) as f:
            self.spec = yaml.safe_load(f)

        self._endpoints: dict[str, dict[str, Any]] = {}
        self._schemas: dict[str, dict[str, Any]] = {}
        self._tags: dict[str, list[str]] = {}

        self._index_endpoints()
        self._index_schemas()

    def _index_endpoints(self) -> None:
        """Index all endpoints by path."""
        paths = self.spec.get("paths", {})
        for path, methods in paths.items():
            self._endpoints[path] = {}
            for method, details in methods.items():
                if method.startswith("x-"):
                    continue
                self._endpoints[path][method] = details

                # Index by tag
                for tag in details.get("tags", []):
                    if tag not in self._tags:
                        self._tags[tag] = []
                    self._tags[tag].append(f"{method.upper()} {path}")

    def _index_schemas(self) -> None:
        """Index all schemas."""
        components = self.spec.get("components", {})
        self._schemas = components.get("schemas", {})

    def find_endpoints(self, pattern: str) -> list[dict[str, Any]]:
        """
        Find endpoints matching a regex pattern.

        Args:
            pattern: Regex pattern to match against path

        Returns:
            List of dicts with path, method, summary, operationId
        """
        results = []
        regex = re.compile(pattern, re.IGNORECASE)

        for path, methods in self._endpoints.items():
            if regex.search(path):
                for method, details in methods.items():
                    results.append({
                        "path": path,
                        "method": method.upper(),
                        "summary": details.get("summary", ""),
                        "operationId": details.get("operationId", ""),
                        "tags": details.get("tags", []),
                    })
        return results

    def get_endpoint(self, path: str, method: str = None) -> dict[str, Any] | None:
        """
        Get full details for an endpoint.

        Args:
            path: Endpoint path (e.g., "/Invoice/{invoiceId}")
            method: HTTP method (optional, returns all if not specified)

        Returns:
            Endpoint details or None if not found
        """
        endpoint = self._endpoints.get(path)
        if not endpoint:
            return None
        if method:
            return endpoint.get(method.lower())
        return endpoint

    def get_endpoint_parameters(self, path: str, method: str = "get") -> list[dict[str, Any]]:
        """Get parameters for an endpoint."""
        endpoint = self.get_endpoint(path, method)
        if not endpoint:
            return []
        return endpoint.get("parameters", [])

    def get_schema(self, name: str) -> dict[str, Any] | None:
        """
        Get a schema by name.

        Args:
            name: Schema name (e.g., "Invoice", "Model_InvoiceResponse")

        Returns:
            Schema definition or None
        """
        # Try exact match first
        if name in self._schemas:
            return self._schemas[name]

        # Try with Model_ prefix
        if f"Model_{name}" in self._schemas:
            return self._schemas[f"Model_{name}"]

        # Try case-insensitive
        for schema_name, schema in self._schemas.items():
            if schema_name.lower() == name.lower():
                return schema

        return None

    def get_schema_properties(self, name: str) -> dict[str, Any]:
        """Get just the properties of a schema."""
        schema = self.get_schema(name)
        if not schema:
            return {}
        return schema.get("properties", {})

    def find_schemas(self, pattern: str) -> list[str]:
        """Find schema names matching a pattern."""
        regex = re.compile(pattern, re.IGNORECASE)
        return [name for name in self._schemas if regex.search(name)]

    def list_tags(self) -> list[str]:
        """List all available tags (resource groups)."""
        return list(self._tags.keys())

    def get_endpoints_by_tag(self, tag: str) -> list[str]:
        """Get all endpoints for a tag."""
        return self._tags.get(tag, [])

    def search(self, keyword: str) -> dict[str, Any]:
        """
        Search everywhere for a keyword.

        Args:
            keyword: Keyword to search for

        Returns:
            Dict with matching endpoints, schemas, and schema_properties
        """
        keyword_lower = keyword.lower()
        results = {
            "endpoints": [],
            "schemas": [],
            "schema_properties": [],
        }

        # Search endpoints
        for path, methods in self._endpoints.items():
            for method, details in methods.items():
                searchable = f"{path} {details.get('summary', '')} {details.get('description', '')} {details.get('operationId', '')}"
                if keyword_lower in searchable.lower():
                    results["endpoints"].append({
                        "path": path,
                        "method": method.upper(),
                        "summary": details.get("summary", ""),
                    })

        # Search schema names
        for name in self._schemas:
            if keyword_lower in name.lower():
                results["schemas"].append(name)

        # Search schema properties
        for schema_name, schema in self._schemas.items():
            props = schema.get("properties", {})
            for prop_name, prop_def in props.items():
                searchable = f"{prop_name} {prop_def.get('description', '')}"
                if keyword_lower in searchable.lower():
                    results["schema_properties"].append({
                        "schema": schema_name,
                        "property": prop_name,
                        "type": prop_def.get("type", ""),
                        "description": prop_def.get("description", "")[:100],
                    })

        return results

    def summary(self) -> dict[str, int]:
        """Get summary statistics about the API."""
        endpoint_count = sum(len(m) for m in self._endpoints.values())
        return {
            "endpoints": endpoint_count,
            "paths": len(self._endpoints),
            "schemas": len(self._schemas),
            "tags": len(self._tags),
        }


def print_results(results: Any, max_items: int = 20) -> None:
    """Pretty print search results."""
    if isinstance(results, dict):
        for key, value in results.items():
            if isinstance(value, list) and value:
                print(f"\n{key.upper()}:")
                for item in value[:max_items]:
                    if isinstance(item, dict):
                        print(f"  - {item}")
                    else:
                        print(f"  - {item}")
                if len(value) > max_items:
                    print(f"  ... and {len(value) - max_items} more")
    elif isinstance(results, list):
        for item in results[:max_items]:
            print(f"  - {item}")
        if len(results) > max_items:
            print(f"  ... and {len(results) - max_items} more")
    else:
        print(results)


if __name__ == "__main__":
    import sys

    spec_path = sys.argv[1] if len(sys.argv) > 1 else "openapi.yaml"
    idx = OpenAPIIndex(spec_path)

    print("OpenAPI Index loaded")
    print(f"Summary: {idx.summary()}")
    print(f"\nTags: {idx.list_tags()}")

    # Interactive mode
    print("\nEnter searches (or 'quit' to exit):")
    while True:
        try:
            query = input("\n> ").strip()
        except EOFError:
            break

        if query.lower() in ("quit", "exit", "q"):
            break

        if query.startswith("/"):
            # Endpoint lookup
            result = idx.get_endpoint(query)
            print_results(result)
        elif query.startswith("schema:"):
            # Schema lookup
            name = query.split(":", 1)[1].strip()
            result = idx.get_schema_properties(name)
            print_results(result)
        elif query.startswith("endpoints:"):
            # Find endpoints
            pattern = query.split(":", 1)[1].strip()
            result = idx.find_endpoints(pattern)
            print_results(result)
        else:
            # General search
            result = idx.search(query)
            print_results(result)