diff --git a/src/core/models.py b/src/core/models.py index 2c6d665..fbff5a0 100644 --- a/src/core/models.py +++ b/src/core/models.py @@ -1,206 +1,192 @@ -"""Pydantic models for OpenAPI specification components.""" +from enum import Enum +from typing import Any, Optional, Union -from typing import Any, Dict, List, Optional, Union -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field -class SchemaProperty(BaseModel): - """A property within a schema object.""" +class HttpMethod(str, Enum): + GET = "get" + POST = "post" + PUT = "put" + DELETE = "delete" + PATCH = "patch" + OPTIONS = "options" + HEAD = "head" - 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 ParameterIn(str, Enum): + PATH = "path" + QUERY = "query" + HEADER = "header" + COOKIE = "cookie" + + +class SchemaType(str, Enum): + STRING = "string" + NUMBER = "number" + INTEGER = "integer" + BOOLEAN = "boolean" + ARRAY = "array" + OBJECT = "object" class Schema(BaseModel): - """A schema definition from the OpenAPI spec.""" + model_config = ConfigDict(populate_by_name=True) - 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 + example: Optional[Any] = None + properties: Optional[dict[str, "Schema"]] = None + items: Optional["Schema"] = None + required: Optional[list[str]] = None + enum: Optional[list[Any]] = None + all_of: Optional[list["Schema"]] = Field(None, alias="allOf") + any_of: Optional[list["Schema"]] = Field(None, alias="anyOf") + one_of: Optional[list["Schema"]] = Field(None, alias="oneOf") + not_: Optional["Schema"] = Field(None, alias="not") + ref: Optional[str] = Field(None, alias="$ref") + additional_properties: Optional[Union[bool, "Schema"]] = Field( + None, alias="additionalProperties" + ) class Parameter(BaseModel): - """A parameter definition for an endpoint.""" + model_config = {"populate_by_name": True} name: str - in_field: str = "query" + in_: ParameterIn = Field(..., alias="in") description: Optional[str] = None required: Optional[bool] = None deprecated: Optional[bool] = None - allow_empty_value: Optional[bool] = None + allow_empty_value: Optional[bool] = Field(None, alias="allowEmptyValue") style: Optional[str] = None explode: Optional[bool] = None - allow_reserved: Optional[bool] = None - schema: Optional[SchemaProperty] = None + allow_reserved: Optional[bool] = Field(None, alias="allowReserved") + schema: Optional[Schema] = Field(None, alias="schema") 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] + examples: Optional[dict[str, Any]] = None 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 + content: Optional[dict[str, Any]] = None + headers: Optional[dict[str, Any]] = None + links: Optional[dict[str, Any]] = None -class Endpoint(BaseModel): - """An API endpoint definition.""" +class RequestBody(BaseModel): + description: Optional[str] = None + required: Optional[bool] = None + content: dict[str, Any] = Field(default_factory=dict) - path: str - method: str - operation_id: Optional[str] = None + +class Operation(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + tags: Optional[list[str]] = None summary: Optional[str] = None description: Optional[str] = None + external_docs: Optional[dict[str, str]] = Field(None, alias="externalDocs") + operation_id: Optional[str] = Field(None, alias="operationId") + parameters: Optional[list[Parameter]] = None + request_body: Optional[RequestBody] = Field(None, alias="requestBody") + responses: dict[str, Response] = Field(default_factory=dict) 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 + security: Optional[list[dict[str, list[str]]]] = None + servers: Optional[list[dict[str, Any]]] = None + + +class PathItem(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + ref: Optional[str] = Field(None, alias="$ref") + summary: Optional[str] = None + description: Optional[str] = None + get: Optional[Operation] = None + put: Optional[Operation] = None + post: Optional[Operation] = None + delete: Optional[Operation] = None + options: Optional[Operation] = None + head: Optional[Operation] = None + patch: Optional[Operation] = None + trace: Optional[Operation] = None + servers: Optional[list[dict[str, Any]]] = None + parameters: Optional[list[Parameter]] = None + + +class Contact(BaseModel): + name: Optional[str] = None + url: Optional[str] = None + email: Optional[str] = None + + +class License(BaseModel): + name: str + url: Optional[str] = None 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 + contact: Optional[Contact] = None + license: Optional[License] = None + + +class ServerVariable(BaseModel): + enum: Optional[list[str]] = None + default: str + description: Optional[str] = None + + +class Server(BaseModel): + url: str + description: Optional[str] = None + variables: Optional[dict[str, ServerVariable]] = None + + +class SecurityScheme(BaseModel): + type: str + scheme: Optional[str] = None + bearer_format: Optional[str] = Field(None, alias="bearerFormat") + flows: Optional[dict[str, Any]] = None + open_id_connect_url: Optional[str] = Field(None, alias="openIdConnectUrl") + description: Optional[str] = None + + +class Components(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + schemas: Optional[dict[str, Schema]] = None + responses: Optional[dict[str, Response]] = None + parameters: Optional[dict[str, Parameter]] = None + request_bodies: Optional[dict[str, RequestBody]] = Field(None, alias="requestBodies") + headers: Optional[dict[str, Any]] = None + security_schemes: Optional[dict[str, SecurityScheme]] = Field( + None, alias="securitySchemes" + ) + links: Optional[dict[str, Any]] = None + callbacks: 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 + external_docs: Optional[dict[str, str]] = Field(None, alias="externalDocs") class OpenAPISpec(BaseModel): - """The complete OpenAPI specification model.""" + model_config = ConfigDict(populate_by_name=True) openapi: str info: Info - paths: Dict[str, Dict[str, Any]] - 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] - - 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"]) - - 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 [] + servers: Optional[list[Server]] = None + paths: dict[str, PathItem] + components: Optional[Components] = None + security: Optional[list[dict[str, list[str]]]] = None + tags: Optional[list[Tag]] = None + external_docs: Optional[dict[str, str]] = Field(None, alias="externalDocs")