gitignore

.gitignore

@@ -0,0 +1,5 @@
+/venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+.env

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

app/main.py

@@ -0,0 +1,16 @@
+from fastapi import FastAPI
+
+from .routes import router
+
+app = FastAPI(
+    title="OpenAPI Parser",
+    description="Recibe una especificación OpenAPI y la transforma en información útil para visualización gráfica de endpoints.",
+    version="0.1.0",
+)
+
+app.include_router(router)
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok"}

app/parser.py

@@ -0,0 +1,140 @@
+import httpx
+
+from .schemas import (
+    EndpointInfo,
+    Parameter,
+    RequestBody,
+    Response,
+)
+
+JSON_TYPE_MAP = {
+    "string": "string",
+    "integer": "integer",
+    "number": "number",
+    "boolean": "boolean",
+    "array": "array",
+    "object": "object",
+}
+
+
+def _resolve_ref(spec: dict, ref: str) -> dict:
+    """Resolve a $ref pointer like '#/components/schemas/MyModel'."""
+    parts = ref.lstrip("#/").split("/")
+    node = spec
+    for part in parts:
+        node = node[part]
+    return node
+
+
+def _get_type(schema: dict, spec: dict) -> str:
+    if "$ref" in schema:
+        schema = _resolve_ref(spec, schema["$ref"])
+    if "type" in schema:
+        return JSON_TYPE_MAP.get(schema["type"], schema["type"])
+    if "anyOf" in schema:
+        types = [_get_type(s, spec) for s in schema["anyOf"] if s.get("type") != "null"]
+        return types[0] if len(types) == 1 else " | ".join(types)
+    return "unknown"
+
+
+def _extract_fields(schema: dict, spec: dict) -> dict[str, str]:
+    """Extract field_name -> type from an object schema."""
+    if "$ref" in schema:
+        schema = _resolve_ref(spec, schema["$ref"])
+    properties = schema.get("properties", {})
+    return {name: _get_type(prop, spec) for name, prop in properties.items()}
+
+
+def _parse_parameters(params: list[dict], spec: dict) -> list[Parameter]:
+    result = []
+    for p in params:
+        if "$ref" in p:
+            p = _resolve_ref(spec, p["$ref"])
+        schema = p.get("schema", {})
+        result.append(
+            Parameter(
+                name=p["name"],
+                location=p["in"],
+                type=_get_type(schema, spec),
+                required=p.get("required", False),
+                description=p.get("description"),
+            )
+        )
+    return result
+
+
+def _parse_request_body(body: dict | None, spec: dict) -> RequestBody | None:
+    if not body:
+        return None
+    if "$ref" in body:
+        body = _resolve_ref(spec, body["$ref"])
+    content = body.get("content", {})
+    for content_type, media in content.items():
+        schema = media.get("schema", {})
+        fields = _extract_fields(schema, spec)
+        return RequestBody(content_type=content_type, fields=fields)
+    return None
+
+
+def _parse_responses(responses: dict, spec: dict) -> list[Response]:
+    result = []
+    for status_code, resp in responses.items():
+        if "$ref" in resp:
+            resp = _resolve_ref(spec, resp["$ref"])
+        content = resp.get("content", {})
+        if content:
+            for content_type, media in content.items():
+                schema = media.get("schema", {})
+                fields = _extract_fields(schema, spec)
+                result.append(
+                    Response(
+                        status_code=str(status_code),
+                        description=resp.get("description"),
+                        content_type=content_type,
+                        fields=fields,
+                    )
+                )
+                break
+        else:
+            result.append(
+                Response(
+                    status_code=str(status_code),
+                    description=resp.get("description"),
+                    fields={},
+                )
+            )
+    return result
+
+
+def parse_endpoint(spec: dict, path: str, method: str, operation: dict) -> EndpointInfo:
+    return EndpointInfo(
+        path=path,
+        method=method.upper(),
+        summary=operation.get("summary"),
+        description=operation.get("description"),
+        operation_id=operation.get("operationId"),
+        parameters=_parse_parameters(operation.get("parameters", []), spec),
+        request_body=_parse_request_body(operation.get("requestBody"), spec),
+        responses=_parse_responses(operation.get("responses", {}), spec),
+    )
+
+
+async def fetch_and_parse(spec_url: str, path: str | None = None, method: str | None = None) -> list[EndpointInfo]:
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(str(spec_url), follow_redirects=True)
+        resp.raise_for_status()
+        spec = resp.json()
+
+    endpoints: list[EndpointInfo] = []
+
+    for ep_path, methods in spec.get("paths", {}).items():
+        if path and ep_path != path:
+            continue
+        for ep_method, operation in methods.items():
+            if ep_method in ("parameters", "summary", "description", "servers"):
+                continue
+            if method and ep_method.upper() != method.upper():
+                continue
+            endpoints.append(parse_endpoint(spec, ep_path, ep_method, operation))
+
+    return endpoints

app/routes.py

@@ -0,0 +1,27 @@
+from fastapi import APIRouter, HTTPException, Query
+
+from .parser import fetch_and_parse
+from .schemas import ParseResponse
+
+router = APIRouter()
+
+
+@router.get("/parse", response_model=ParseResponse)
+async def parse_spec(
+    spec_url: str = Query(..., description="URL del openapi.json a parsear"),
+    path: str | None = Query(None, description="Path específico, ej: /whatsapp"),
+    method: str | None = Query(None, description="Método HTTP, ej: GET, POST"),
+):
+    try:
+        endpoints = await fetch_and_parse(
+            spec_url=spec_url,
+            path=path,
+            method=method,
+        )
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    if not endpoints:
+        raise HTTPException(status_code=404, detail="No matching endpoints found")
+
+    return ParseResponse(endpoints=endpoints)

app/schemas.py

@@ -0,0 +1,42 @@
+from pydantic import BaseModel, HttpUrl
+
+
+class ParseRequest(BaseModel):
+    spec_url: HttpUrl
+    path: str | None = None
+    method: str | None = None
+
+
+class Parameter(BaseModel):
+    name: str
+    location: str  # query, path, header, cookie
+    type: str
+    required: bool
+    description: str | None = None
+
+
+class RequestBody(BaseModel):
+    content_type: str
+    fields: dict[str, str]  # field_name -> type
+
+
+class Response(BaseModel):
+    status_code: str
+    description: str | None = None
+    content_type: str | None = None
+    fields: dict[str, str]  # field_name -> type
+
+
+class EndpointInfo(BaseModel):
+    path: str
+    method: str
+    summary: str | None = None
+    description: str | None = None
+    operation_id: str | None = None
+    parameters: list[Parameter]
+    request_body: RequestBody | None = None
+    responses: list[Response]
+
+
+class ParseResponse(BaseModel):
+    endpoints: list[EndpointInfo]

requirements.txt

@@ -0,0 +1,2 @@
+fastapi[standard]>=0.115.0
+httpx>=0.27.0