This commit is contained in:
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user