diff --git a/src/core/models.py b/src/core/models.py index c928f37..2c6d665 100644 --- a/src/core/models.py +++ b/src/core/models.py @@ -1,48 +1,206 @@ +"""Pydantic models for OpenAPI specification components.""" + +from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field -from typing import Optional, List, Dict, Any -from enum import Enum -class HTTPMethod(str, Enum): - GET = "GET" - POST = "PUT" - PUT = "PUT" - DELETE = "DELETE" - PATCH = "PATCH" - OPTIONS = "OPTIONS" - HEAD = "HEAD" +class SchemaProperty(BaseModel): + """A property within a schema object.""" + + type: Optional[str] = None + format: Optional[str] = None + description: Optional[str] = None + required: Optional[bool] = None + nullable: Optional[bool] = None + default: Optional[Any] = None + example: Optional[Any] = None + examples: Optional[List[Any]] = None + minimum: Optional[float] = None + maximum: Optional[float] = None + min_length: Optional[int] = None + max_length: Optional[int] = None + pattern: Optional[str] = None + enum: Optional[List[Any]] = None + items: Optional["SchemaProperty"] = None + properties: Optional[Dict[str, "SchemaProperty"]] = None + additional_properties: Optional[Union[bool, "SchemaProperty"]] = None + all_of: Optional[List["SchemaProperty"]] = None + one_of: Optional[List["SchemaProperty"]] = None + any_of: Optional[List["SchemaProperty"]] = None + not_: Optional["SchemaProperty"] = Field(None, alias="not") + ref: Optional[str] = Field(None, alias="$ref") + read_only: Optional[bool] = None + write_only: Optional[bool] = None + + class Config: + populate_by_name = True + + +class Schema(BaseModel): + """A schema definition from the OpenAPI spec.""" + + name: str + ref: Optional[str] = None + type: Optional[str] = None + format: Optional[str] = None + description: Optional[str] = None + required: Optional[List[str]] = None + nullable: Optional[bool] = None + deprecated: Optional[bool] = None + properties: Optional[Dict[str, SchemaProperty]] = None + additional_properties: Optional[Union[bool, SchemaProperty]] = None + all_of: Optional[List[SchemaProperty]] = None + one_of: Optional[List[SchemaProperty]] = None + any_of: Optional[List[SchemaProperty]] = None + not_: Optional[SchemaProperty] = Field(None, alias="not") + items: Optional[SchemaProperty] = None + enum: Optional[List[Any]] = None + example: Optional[Any] = None + examples: Optional[List[Any]] = None + default: Optional[Any] = None + + class Config: + populate_by_name = True + + +class Parameter(BaseModel): + """A parameter definition for an endpoint.""" + + name: str + in_field: str = "query" + description: Optional[str] = None + required: Optional[bool] = None + deprecated: Optional[bool] = None + allow_empty_value: Optional[bool] = None + style: Optional[str] = None + explode: Optional[bool] = None + allow_reserved: Optional[bool] = None + schema: Optional[SchemaProperty] = None + example: Optional[Any] = None + examples: Optional[Dict[str, Any]] = None + content: Optional[Dict[str, Any]] = None + + +class MediaType(BaseModel): + """A media type object for request/response bodies.""" + + schema: Optional[SchemaProperty] = None + example: Optional[Any] = None + examples: Optional[Dict[str, Any]] = None + encoding: Optional[Dict[str, Any]] = None + + +class RequestBody(BaseModel): + """A request body definition.""" + + description: Optional[str] = None + required: Optional[bool] = None + content: Dict[str, MediaType] + + +class Response(BaseModel): + """A response definition for an endpoint.""" + + description: str + headers: Optional[Dict[str, Any]] = None + content: Optional[Dict[str, MediaType]] = None + links: Optional[Dict[str, Any]] = None class Endpoint(BaseModel): + """An API endpoint definition.""" + path: str method: str + operation_id: Optional[str] = None summary: Optional[str] = None description: Optional[str] = None - operation_id: Optional[str] = None - tags: List[str] = Field(default_factory=list) - parameters: List[Dict[str, Any]] = Field(default_factory=list) - request_body: Optional[Dict[str, Any]] = None - responses: Dict[str, Any] = Field(default_factory=dict) - deprecated: bool = False + deprecated: Optional[bool] = None + tags: Optional[List[str]] = None + parameters: Optional[List[Parameter]] = None + requestBody: Optional[RequestBody] = None + responses: Dict[str, Response] + security: Optional[List[Dict[str, List[str]]]] = None + externalDocs: Optional[Dict[str, str]] = None -class APISpec(BaseModel): +class Info(BaseModel): + """The info object from the OpenAPI spec.""" + + title: str + version: str + description: Optional[str] = None + terms_of_service: Optional[str] = Field(None, alias="termsOfService") + contact: Optional[Dict[str, Any]] = None + license: Optional[Dict[str, Any]] = None + + +class Tag(BaseModel): + """A tag definition for organizing endpoints.""" + + name: str + description: Optional[str] = None + external_docs: Optional[Dict[str, str]] = Field(None, alias="externalDocs") + + class Config: + populate_by_name = True + + +class OpenAPISpec(BaseModel): + """The complete OpenAPI specification model.""" + openapi: str - info: Dict[str, Any] + info: Info paths: Dict[str, Dict[str, Any]] - tags: List[Dict[str, str]] = Field(default_factory=list) - servers: List[Dict[str, str]] = Field(default_factory=list) - components: Dict[str, Any] = Field(default_factory=dict) + servers: Optional[List[Dict[str, str]]] = None + components: Optional[Dict[str, Any]] = None + security: Optional[List[Dict[str, List[str]]]] = None + tags: Optional[List[Tag]] = None + external_docs: Optional[Dict[str, str]] = Field(None, alias="externalDocs") + def get_endpoints(self) -> List[Endpoint]: + """Extract all endpoints from the spec.""" + endpoints = [] + for path, methods in self.paths.items(): + for method, details in methods.items(): + if method in ["get", "post", "put", "patch", "delete", "options", "head", "trace"]: + params = details.get("parameters", []) + parameters = [Parameter(**p) if isinstance(p, dict) else p for p in params] -class RequestExample(BaseModel): - method: str - path: str - headers: Dict[str, str] = Field(default_factory=dict) - body: Optional[Any] = None + responses = {} + for status_code, response_def in details.get("responses", {}).items(): + responses[status_code] = Response(**response_def) + request_body = None + if "requestBody" in details: + request_body = RequestBody(**details["requestBody"]) -class ResponseExample(BaseModel): - status_code: int - headers: Dict[str, str] = Field(default_factory=dict) - body: Optional[Any] = None + endpoint = Endpoint( + path=path, + method=method.upper(), + operation_id=details.get("operationId"), + summary=details.get("summary"), + description=details.get("description"), + deprecated=details.get("deprecated", False), + tags=details.get("tags"), + parameters=parameters, + responses=responses, + security=details.get("security"), + requestBody=request_body, + externalDocs=details.get("externalDocs"), + ) + endpoints.append(endpoint) + return endpoints + + def get_schemas(self) -> Dict[str, Schema]: + """Extract all schemas from the components section.""" + schemas = {} + components = self.components or {} + for schema_name, schema_def in components.get("schemas", {}).items(): + schema = Schema(name=schema_name, **schema_def) + schemas[schema_name] = schema + return schemas + + def get_tags(self) -> List[Tag]: + """Get all tags defined in the spec.""" + return self.tags or []