Moibe/openapi-parser

1import httpx2 3from .schemas import (4    EndpointInfo,5    FieldInfo,6    Parameter,7    RequestBody,8    Response,9)10 11JSON_TYPE_MAP = {12    "string": "string",13    "integer": "integer",14    "number": "number",15    "boolean": "boolean",16    "array": "array",17    "object": "object",18}19 20 21def _resolve_ref(spec: dict, ref: str) -> dict:22    """Resolve a $ref pointer like '#/components/schemas/MyModel'."""23    parts = ref.lstrip("#/").split("/")24    node = spec25    for part in parts:26        node = node[part]27    return node28 29 30def _get_type(schema: dict, spec: dict) -> str:31    if "$ref" in schema:32        schema = _resolve_ref(spec, schema["$ref"])33    if "type" in schema:34        return JSON_TYPE_MAP.get(schema["type"], schema["type"])35    if "anyOf" in schema:36        types = [_get_type(s, spec) for s in schema["anyOf"] if s.get("type") != "null"]37        return types[0] if len(types) == 1 else " | ".join(types)38    return "unknown"39 40 41def _get_format(schema: dict, spec: dict) -> str | None:42    if "$ref" in schema:43        schema = _resolve_ref(spec, schema["$ref"])44    return schema.get("format")45 46 47def _extract_fields(schema: dict, spec: dict) -> dict[str, FieldInfo]:48    """Extract field_name -> FieldInfo from an object schema."""49    if "$ref" in schema:50        schema = _resolve_ref(spec, schema["$ref"])51    properties = schema.get("properties", {})52    return {53        name: FieldInfo(type=_get_type(prop, spec), format=_get_format(prop, spec))54        for name, prop in properties.items()55    }56 57 58def _parse_parameters(params: list[dict], spec: dict) -> list[Parameter]:59    result = []60    for p in params:61        if "$ref" in p:62            p = _resolve_ref(spec, p["$ref"])63        schema = p.get("schema", {})64        result.append(65            Parameter(66                name=p["name"],67                location=p["in"],68                type=_get_type(schema, spec),69                required=p.get("required", False),70                description=p.get("description"),71            )72        )73    return result74 75 76def _parse_request_body(body: dict | None, spec: dict) -> RequestBody | None:77    if not body:78        return None79    if "$ref" in body:80        body = _resolve_ref(spec, body["$ref"])81    content = body.get("content", {})82    for content_type, media in content.items():83        schema = media.get("schema", {})84        fields = _extract_fields(schema, spec)85        return RequestBody(content_type=content_type, fields=fields)86    return None87 88 89def _parse_responses(responses: dict, spec: dict) -> list[Response]:90    result = []91    for status_code, resp in responses.items():92        if "$ref" in resp:93            resp = _resolve_ref(spec, resp["$ref"])94        content = resp.get("content", {})95        if content:96            for content_type, media in content.items():97                schema = media.get("schema", {})98                fields = _extract_fields(schema, spec)99                result.append(100                    Response(101                        status_code=str(status_code),102                        description=resp.get("description"),103                        content_type=content_type,104                        fields=fields,105                    )106                )107                break108        else:109            result.append(110                Response(111                    status_code=str(status_code),112                    description=resp.get("description"),113                    fields={},114                )115            )116    return result117 118 119def parse_endpoint(spec: dict, path: str, method: str, operation: dict) -> EndpointInfo:120    return EndpointInfo(121        path=path,122        method=method.upper(),123        summary=operation.get("summary"),124        description=operation.get("description"),125        operation_id=operation.get("operationId"),126        parameters=_parse_parameters(operation.get("parameters", []), spec),127        request_body=_parse_request_body(operation.get("requestBody"), spec),128        responses=_parse_responses(operation.get("responses", {}), spec),129    )130 131 132def _normalize_path(p: str) -> str:133    """Strip trailing slashes for consistent comparison, but keep root '/'."""134    return p.rstrip("/") or "/"135 136 137async def fetch_and_parse(spec_url: str, path: str | None = None, method: str | None = None) -> list[EndpointInfo]:138    async with httpx.AsyncClient() as client:139        resp = await client.get(str(spec_url), follow_redirects=True)140        resp.raise_for_status()141        spec = resp.json()142 143    normalized_path = _normalize_path(path) if path else None144    endpoints: list[EndpointInfo] = []145 146    for ep_path, methods in spec.get("paths", {}).items():147        if normalized_path and _normalize_path(ep_path) != normalized_path:148            continue149        for ep_method, operation in methods.items():150            if ep_method in ("parameters", "summary", "description", "servers"):151                continue152            if method and ep_method.upper() != method.upper():153                continue154            endpoints.append(parse_endpoint(spec, ep_path, ep_method, operation))155 156    return endpoints157