codegen.py

#!/usr/bin/env python3
"""
Generate factory.py from OpenAPI spec.

Reads specs/openapi.json and auto-generates src/roxy_sdk/factory.py with all
domain namespace classes, Roxy aggregate, and create_roxy() factory.

The OpenAPI spec is the only source of truth for *which* tags exist and what
endpoints belong to each. This file only configures how tag names map to
public SDK namespaces:

 1. Most tags derive their snake_case attribute and PascalCase class name
 automatically — e.g. "Vedic Astrology" -> vedic_astrology / VedicAstrologyDomain,
 "Numerology" -> numerology / NumerologyDomain, "Languages" -> languages /
 LanguagesDomain. New tags need NO change here.
 2. A handful of tags use a curated short-form for branding — the public SDK
 exposes ``roxy.astrology``, not ``roxy.western_astrology``. Those overrides
 live in NAMESPACE_ALIASES below.

Adding a brand-new tag with a non-default short-form (e.g. "Crystals and Healing
Stones" -> ``crystals``) is the only reason to touch this file. Otherwise
codegen handles it.

Run: python codegen.py
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

SPEC_PATH = Path("specs/openapi.json")
OUTPUT_PATH = Path("src/roxy_sdk/factory.py")

# Curated short-form namespaces for product branding. Keys are exact tag names
# from the OpenAPI spec; values are the desired snake_case attribute name on
# the Roxy aggregate. The class name is derived from the attribute name (see
# tag_to_class). Any tag not listed here uses snake_case(tag_name) automatically.
NAMESPACE_ALIASES: dict[str, str] = {
 "Western Astrology": "astrology",
 "Crystals and Healing Stones": "crystals",
 "Location and Timezone": "location",
 "I-Ching": "iching",
}


def _snake_case(name: str) -> str:
 """Split on non-alphanumerics, lowercase, join with underscores."""
 parts = [p for p in re.split(r"[^a-zA-Z0-9]+", str(name)) if p]
 if not parts:
 return str(name).lower()
 return "_".join(p.lower() for p in parts)


def tag_to_attr(name: str) -> str:
 """Return the snake_case attribute name on the Roxy aggregate for a tag."""
 return NAMESPACE_ALIASES.get(name, _snake_case(name))


def tag_to_class(name: str) -> str:
 """Return the PascalCase domain class name for a tag, suffixed with Domain."""
 attr = tag_to_attr(name)
 pascal = attr.title().replace("_", "")
 return f"{pascal}Domain"


def tag_summary(tag: dict[str, Any]) -> str:
 """Extract a short, dev-facing summary from the spec's tag object.

 Prefers ``x-sdk-summary`` if present (forward-compat for when the server adds
 explicit SDK summaries per tag). Otherwise takes the first sentence of
 ``description``, capped at ~120 chars.
 """
 sdk_summary = tag.get("x-sdk-summary")
 if isinstance(sdk_summary, str) and sdk_summary.strip():
 return sdk_summary.strip()
 desc = (tag.get("description") or "").strip()
 if not desc:
 return str(tag.get("name") or "")
 first_sentence = re.split(r"\.\s+", desc, maxsplit=1)[0].strip()
 flat = re.sub(r"\s+", " ", first_sentence)
 return flat[:117].rstrip() + "..." if len(flat) > 120 else flat


def camel_to_snake(name: str) -> str:
 s1 = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", name)
 return re.sub(r"([a-z\d])([A-Z])", r"\1_\2", s1).lower()


def json_type_to_python(schema: dict) -> str:
 t = schema.get("type", "string")
 if t == "integer":
 return "int"
 if t == "number":
 return "float"
 if t == "boolean":
 return "bool"
 if t == "array":
 inner = json_type_to_python(schema.get("items", {}))
 return f"list[{inner}]"
 if t == "object":
 return "dict[str, Any]"
 return "str"


def resolve_ref(spec: dict, ref: str) -> dict:
 parts = ref.lstrip("#/").split("/")
 obj = spec
 for part in parts:
 obj = obj[part]
 return obj


def extract_body(spec: dict, operation: dict) -> tuple[dict, list[str]]:
 """Extract request body properties and required fields."""
 rb = operation.get("requestBody", {})
 if not rb:
 return {}, []
 content = rb.get("content", {}).get("application/json", {})
 schema = content.get("schema", {})
 if "$ref" in schema:
 schema = resolve_ref(spec, schema["$ref"])
 props = {}
 for name, prop in schema.get("properties", {}).items():
 if "$ref" in prop:
 prop = resolve_ref(spec, prop["$ref"])
 props[name] = prop
 return props, schema.get("required", [])


def build_method(op: dict) -> str:
 """Generate sync + async wrapper methods for one endpoint."""
 func = camel_to_snake(op["operationId"])
 http = op["method"]
 path = op["path"]
 summary = (op.get("summary") or func).replace("'", "").replace('"', "")

 path_params = [p["name"] for p in op.get("parameters", []) if p.get("in") == "path"]
 query_params = [p for p in op.get("parameters", []) if p.get("in") == "query"]

 # Build method signature params
 sig_parts: list[str] = []
 for p in path_params:
 sig_parts.append(f"{camel_to_snake(p)}: str")

 body_props = op.get("body_properties", {})
 body_required = op.get("body_required_fields", [])
 if body_props:
 for prop_name, prop_info in body_props.items():
 py_name = camel_to_snake(prop_name)
 py_type = json_type_to_python(prop_info)
 if prop_name in body_required:
 sig_parts.append(f"{py_name}: {py_type}")
 else:
 sig_parts.append(f"{py_name}: {py_type} | None = None")

 # Query params (skip lang, add at end)
 for qp in query_params:
 if qp["name"] == "lang":
 continue
 py_name = camel_to_snake(qp["name"])
 py_type = json_type_to_python(qp.get("schema", {}))
 sig_parts.append(f"{py_name}: {py_type} | None = None")

 has_lang = any(qp["name"] == "lang" for qp in query_params)
 if has_lang:
 sig_parts.append("lang: str | None = None")

 sig = ", ".join(["self", "*"] + sig_parts) if sig_parts else "self"

 # Build f-string path
 fpath = path
 for p in path_params:
 fpath = fpath.replace(f"{{{p}}}", f"{{{camel_to_snake(p)}}}")

 # Build method body
 body_lines: list[str] = []

 if http == "get":
 if query_params:
 body_lines.append(" params: dict[str, Any] = {}")
 for qp in query_params:
 py_name = camel_to_snake(qp["name"])
 body_lines.append(f" if {py_name} is not None:")
 body_lines.append(f' params["{qp["name"]}"] = {py_name}')
 body_lines.append(f' return self._get(f"{fpath}", params=params or None)')
 else:
 body_lines.append(f' return self._get(f"{fpath}")')
 else: # POST
 if body_props:
 body_lines.append(" body: dict[str, Any] = {}")
 for prop_name in body_props:
 py_name = camel_to_snake(prop_name)
 if prop_name in body_required:
 body_lines.append(f' body["{prop_name}"] = {py_name}')
 else:
 body_lines.append(f" if {py_name} is not None:")
 body_lines.append(f' body["{prop_name}"] = {py_name}')
 else:
 body_lines.append(" body: dict[str, Any] = {}")

 if has_lang:
 body_lines.append(" params: dict[str, Any] = {}")
 body_lines.append(" if lang is not None:")
 body_lines.append(' params["lang"] = lang')
 body_lines.append(f' return self._post(f"{fpath}", body, params=params or None)')
 else:
 body_lines.append(f' return self._post(f"{fpath}", body)')

 body_code = "\n".join(body_lines)
 async_body = body_code.replace("self._get(", "await self._get_async(").replace(
 "self._post(", "await self._post_async("
 )

 return f""" def {func}({sig}) -> Any:
 \"\"\"{summary}\"\"\"
{body_code}

 async def {func}_async({sig}) -> Any:
 \"\"\"{summary} (async)\"\"\"
{async_body}
"""


def main() -> None:
 spec = json.loads(SPEC_PATH.read_text())

 # Index tag objects (with descriptions) for summary extraction.
 tag_objects: dict[str, dict[str, Any]] = {
 t["name"]: t for t in spec.get("tags", []) if isinstance(t, dict) and "name" in t
 }

 # Warn on stale NAMESPACE_ALIASES — aliases pointing at tags no longer in
 # the spec. New tags are fine; they auto-derive via tag_to_attr / tag_to_class.
 spec_tag_names: set[str] = set(tag_objects.keys())
 for _, path_item in spec.get("paths", {}).items():
 for http_method in ("get", "post"):
 op = path_item.get(http_method)
 if op:
 spec_tag_names.add((op.get("tags", ["Other"]) or ["Other"])[0])
 for alias_tag in NAMESPACE_ALIASES:
 if alias_tag not in spec_tag_names:
 print(
 f"WARNING: NAMESPACE_ALIASES entry '{alias_tag}' is stale — "
 "no such tag in the OpenAPI spec. Remove it from codegen.py."
 )

 # Group operations by tag
 domains: dict[str, list[dict]] = {}
 for path, path_item in spec.get("paths", {}).items():
 for http_method in ("get", "post"):
 operation = path_item.get(http_method)
 if not operation:
 continue
 tags = operation.get("tags", ["Other"])
 tag = tags[0]
 oid = operation.get("operationId", "")
 if not oid:
 continue

 body_props, body_req = extract_body(spec, operation)

 if tag not in domains:
 domains[tag] = []
 domains[tag].append(
 {
 "operationId": oid,
 "method": http_method,
 "path": path,
 "summary": operation.get("summary", ""),
 "parameters": operation.get("parameters", []),
 "body_properties": body_props,
 "body_required_fields": body_req,
 }
 )

 # Write output
 out: list[str] = [
 '"""',
 "AUTO-GENERATED by codegen.py from OpenAPI spec. Do not edit manually.",
 "Regenerate with: python generate.py",
 '"""',
 "from __future__ import annotations",
 "",
 "from typing import Any",
 "",
 "import httpx",
 "",
 "from roxy_sdk.version import VERSION",
 "",
 "",
 '_BASE_URL = "https://roxyapi.com/api/v2"',
 "",
 "",
 "class RoxyAPIError(Exception):",
 ' """Error returned by the RoxyAPI."""',
 "",
 " def __init__(self, error: str, code: str, status_code: int) -> None:",
 " self.error = error",
 " self.code = code",
 " self.status_code = status_code",
 ' super().__init__(f"[{status_code}] {code}: {error}")',
 "",
 " def __repr__(self) -> str:",
 ' return f"RoxyAPIError(error={self.error!r}, code={self.code!r}, status_code={self.status_code})"',
 "",
 "",
 "def _default_headers(api_key: str) -> dict[str, str]:",
 " return {",
 ' "X-API-Key": api_key,',
 ' "X-SDK-Client": f"roxy-sdk-python/{VERSION}",',
 ' "Accept": "application/json",',
 " }",
 "",
 "",
 "def _handle_response(resp: httpx.Response) -> Any:",
 ' """Parse response, raising RoxyAPIError on 4xx/5xx."""',
 " if resp.status_code >= 400:",
 " try:",
 " body = resp.json()",
 " except Exception:",
 " body = {}",
 " raise RoxyAPIError(",
 ' error=body.get("error", resp.text or "Unknown error"),',
 ' code=body.get("code", "unknown"),',
 " status_code=resp.status_code,",
 " )",
 " return resp.json()",
 "",
 "",
 "class _BaseDomain:",
 ' """Base class for all domain namespaces."""',
 "",
 " def __init__(self, client: httpx.Client, async_client: httpx.AsyncClient) -> None:",
 " self._client = client",
 " self._async_client = async_client",
 "",
 " def _get(self, path: str, params: dict[str, Any] | None = None) -> Any:",
 " return _handle_response(self._client.get(path, params=params))",
 "",
 " async def _get_async(self, path: str, params: dict[str, Any] | None = None) -> Any:",
 " return _handle_response(await self._async_client.get(path, params=params))",
 "",
 " def _post(self, path: str, body: dict[str, Any] | None = None, params: dict[str, Any] | None = None) -> Any:",
 " return _handle_response(self._client.post(path, json=body, params=params))",
 "",
 " async def _post_async(self, path: str, body: dict[str, Any] | None = None, params: dict[str, Any] | None = None) -> Any:",
 " return _handle_response(await self._async_client.post(path, json=body, params=params))",
 "",
 ]

 # Domain classes. Emit in spec tag order when available; fall back to
 # alphabetical so brand-new endpoints get a deterministic position.
 spec_tag_order = [
 t["name"] for t in spec.get("tags", []) if isinstance(t, dict) and "name" in t
 ]
 ordered_tags = [t for t in spec_tag_order if t in domains]
 ordered_tags += sorted(t for t in domains if t not in spec_tag_order)

 domain_classes: list[tuple[str, str]] = []
 for tag in ordered_tags:
 class_name = tag_to_class(tag)
 attr_name = tag_to_attr(tag)
 summary = tag_summary(tag_objects.get(tag, {"name": tag}))
 docstring = summary or f"{tag} endpoints."
 domain_classes.append((class_name, attr_name))
 out.append("")
 out.append(f"class {class_name}(_BaseDomain):")
 out.append(f' """{docstring}"""')
 out.append("")
 for op in sorted(domains[tag], key=lambda x: x["operationId"]):
 out.append(build_method(op))

 # Roxy class
 out.append("")
 out.append("class Roxy:")
 out.append(' """')
 out.append(" RoxyAPI Python SDK.")
 out.append(" Reuses HTTP connections. Supports context manager.")
 out.append("")
 out.append(" Usage::")
 out.append(" from roxy_sdk import create_roxy")
 out.append(' roxy = create_roxy("your-api-key")')
 out.append(' horoscope = roxy.astrology.get_daily_horoscope(sign="aries")')
 out.append(' """')
 out.append("")
 out.append(
 " def __init__(self, api_key: str, base_url: str = _BASE_URL, timeout: float = 30.0) -> None:"
 )
 out.append(" if not api_key:")
 out.append(
 ' raise ValueError("API key is required. Get one at https://roxyapi.com/pricing")'
 )
 out.append(" headers = _default_headers(api_key)")
 out.append(
 " self._client = httpx.Client(base_url=base_url, headers=headers, timeout=timeout)"
 )
 out.append(
 " self._async_client = httpx.AsyncClient(base_url=base_url, headers=headers, timeout=timeout)"
 )
 for cn, an in domain_classes:
 out.append(f" self.{an} = {cn}(self._client, self._async_client)")
 out.append("")
 out.append(" def close(self) -> None:")
 out.append(' """Close sync HTTP connections."""')
 out.append(" self._client.close()")
 out.append("")
 out.append(" async def aclose(self) -> None:")
 out.append(' """Close all HTTP connections (sync + async)."""')
 out.append(" self._client.close()")
 out.append(" await self._async_client.aclose()")
 out.append("")
 out.append(" def __enter__(self) -> Roxy:")
 out.append(" return self")
 out.append("")
 out.append(" def __exit__(self, *args: Any) -> None:")
 out.append(" self.close()")
 out.append("")
 out.append(" async def __aenter__(self) -> Roxy:")
 out.append(" return self")
 out.append("")
 out.append(" async def __aexit__(self, *args: Any) -> None:")
 out.append(" await self.aclose()")
 out.append("")
 out.append(" def __repr__(self) -> str:")
 out.append(' return f"Roxy(base_url={self._client.base_url!r})"')
 out.append("")
 out.append("")

 # create_roxy
 out.append(
 "def create_roxy(api_key: str, base_url: str = _BASE_URL, timeout: float = 30.0) -> Roxy:"
 )
 out.append(' """')
 out.append(" Create a configured Roxy instance.")
 out.append("")
 out.append(" Args:")
 out.append(" api_key: Your RoxyAPI key. Get one at https://roxyapi.com/pricing")
 out.append(
 " base_url: Override the default API base URL. Default: https://roxyapi.com/api/v2"
 )
 out.append(" timeout: Request timeout in seconds. Default: 30.")
 out.append(' """')
 out.append(" return Roxy(api_key=api_key, base_url=base_url, timeout=timeout)")
 out.append("")

 OUTPUT_PATH.write_text("\n".join(out))
 print(f"Generated {OUTPUT_PATH} ({len(out)} lines)")


if __name__ == "__main__":
 main()