mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge pull request #147 from jlowin/openapi
Support openapi 3.0 and 3.1
This commit is contained in:
Jeremiah Lowin
GitHub
commit
e119d65273
3 changed files with 1002 additions and 292 deletions
|
|
@ -7,6 +7,8 @@ icon: code-branch
|
|||
|
||||
If you have existing REST APIs documented with the OpenAPI Specification (OAS), FastMCP can automatically generate MCP tools, resources, and resource templates directly from that specification. This provides a quick way to make your existing HTTP APIs accessible to MCP clients and LLMs.
|
||||
|
||||
FastMCP supports both OpenAPI 3.0 and 3.1 specifications for maximum compatibility with existing API definitions.
|
||||
|
||||
## The Goal: API -> MCP Server
|
||||
|
||||
The core idea is to map OpenAPI paths and operations (like `GET /users/{id}` or `POST /orders`) to their corresponding MCP components:
|
||||
|
|
|
|||
|
|
@ -14,6 +14,16 @@ from openapi_pydantic import (
|
|||
Response,
|
||||
Schema,
|
||||
)
|
||||
|
||||
# Import OpenAPI 3.0 models as well
|
||||
from openapi_pydantic.v3.v3_0 import OpenAPI as OpenAPI_30
|
||||
from openapi_pydantic.v3.v3_0 import Operation as Operation_30
|
||||
from openapi_pydantic.v3.v3_0 import Parameter as Parameter_30
|
||||
from openapi_pydantic.v3.v3_0 import PathItem as PathItem_30
|
||||
from openapi_pydantic.v3.v3_0 import Reference as Reference_30
|
||||
from openapi_pydantic.v3.v3_0 import RequestBody as RequestBody_30
|
||||
from openapi_pydantic.v3.v3_0 import Response as Response_30
|
||||
from openapi_pydantic.v3.v3_0 import Schema as Schema_30
|
||||
from pydantic import BaseModel, Field, ValidationError
|
||||
|
||||
from fastmcp.utilities import openapi
|
||||
|
|
@ -176,131 +186,6 @@ def _convert_to_parameter_location(param_in: str) -> ParameterLocation:
|
|||
return "query"
|
||||
|
||||
|
||||
def _extract_parameters(
|
||||
operation_params: list[Parameter | Reference] | None,
|
||||
path_item_params: list[Parameter | Reference] | None,
|
||||
openapi: OpenAPI,
|
||||
) -> list[ParameterInfo]:
|
||||
"""Extracts and resolves parameters using corrected attribute names."""
|
||||
extracted_params: list[ParameterInfo] = []
|
||||
seen_params: dict[
|
||||
tuple[str, str], bool
|
||||
] = {} # Use string keys to avoid type issues
|
||||
all_params_refs = (operation_params or []) + (path_item_params or [])
|
||||
|
||||
for param_or_ref in all_params_refs:
|
||||
try:
|
||||
parameter = cast(Parameter, _resolve_ref(param_or_ref, openapi))
|
||||
if not isinstance(parameter, Parameter):
|
||||
# ... (error logging remains the same)
|
||||
continue
|
||||
|
||||
# --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
|
||||
param_in = parameter.param_in # CORRECTED: Use 'param_in'
|
||||
param_location = _convert_to_parameter_location(param_in)
|
||||
param_schema_obj = parameter.param_schema # CORRECTED: Use 'param_schema'
|
||||
# --- *** ---
|
||||
|
||||
param_key = (parameter.name, param_in)
|
||||
if param_key in seen_params:
|
||||
continue
|
||||
seen_params[param_key] = True
|
||||
|
||||
param_schema_dict = {}
|
||||
if param_schema_obj: # Check if schema exists
|
||||
param_schema_dict = _extract_schema_as_dict(param_schema_obj, openapi)
|
||||
elif parameter.content:
|
||||
# Handle complex parameters with 'content'
|
||||
first_media_type = next(iter(parameter.content.values()), None)
|
||||
if (
|
||||
first_media_type and first_media_type.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
param_schema_dict = _extract_schema_as_dict(
|
||||
first_media_type.media_type_schema, openapi
|
||||
)
|
||||
logger.debug(
|
||||
f"Parameter '{parameter.name}' using schema from 'content' field."
|
||||
)
|
||||
|
||||
# Manually create ParameterInfo instance using correct field names
|
||||
param_info = ParameterInfo(
|
||||
name=parameter.name,
|
||||
location=param_location, # Use converted parameter location
|
||||
required=parameter.required,
|
||||
schema=param_schema_dict, # Populate 'schema' field in IR
|
||||
description=parameter.description,
|
||||
)
|
||||
extracted_params.append(param_info)
|
||||
|
||||
except (
|
||||
ValidationError,
|
||||
ValueError,
|
||||
AttributeError,
|
||||
TypeError,
|
||||
) as e: # Added TypeError
|
||||
param_name = getattr(
|
||||
param_or_ref, "name", getattr(param_or_ref, "ref", "unknown")
|
||||
)
|
||||
logger.error(
|
||||
f"Failed to extract parameter '{param_name}': {e}", exc_info=False
|
||||
)
|
||||
|
||||
return extracted_params
|
||||
|
||||
|
||||
def _extract_request_body(
|
||||
request_body_or_ref: RequestBody | Reference | None, openapi: OpenAPI
|
||||
) -> RequestBodyInfo | None:
|
||||
"""Extracts and resolves the request body using corrected attribute names."""
|
||||
if not request_body_or_ref:
|
||||
return None
|
||||
try:
|
||||
request_body = cast(RequestBody, _resolve_ref(request_body_or_ref, openapi))
|
||||
if not isinstance(request_body, RequestBody):
|
||||
# ... (error logging remains the same)
|
||||
return None
|
||||
|
||||
content_schemas: dict[str, JsonSchema] = {}
|
||||
if request_body.content:
|
||||
for media_type_str, media_type_obj in request_body.content.items():
|
||||
# --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
|
||||
if (
|
||||
isinstance(media_type_obj, MediaType)
|
||||
and media_type_obj.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
# --- *** ---
|
||||
try:
|
||||
# Use the corrected attribute here as well
|
||||
schema_dict = _extract_schema_as_dict(
|
||||
media_type_obj.media_type_schema, openapi
|
||||
)
|
||||
content_schemas[media_type_str] = schema_dict
|
||||
except ValueError as schema_err:
|
||||
logger.error(
|
||||
f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}"
|
||||
)
|
||||
elif not isinstance(media_type_obj, MediaType):
|
||||
logger.warning(
|
||||
f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body."
|
||||
)
|
||||
elif not media_type_obj.media_type_schema: # Corrected check
|
||||
logger.warning(
|
||||
f"Skipping media type '{media_type_str}' in request body because it lacks a schema."
|
||||
)
|
||||
|
||||
return RequestBodyInfo(
|
||||
required=request_body.required,
|
||||
content_schema=content_schemas,
|
||||
description=request_body.description,
|
||||
)
|
||||
except (ValidationError, ValueError, AttributeError) as e:
|
||||
ref_name = getattr(request_body_or_ref, "ref", "unknown")
|
||||
logger.error(
|
||||
f"Failed to extract request body '{ref_name}': {e}", exc_info=False
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _extract_responses(
|
||||
operation_responses: dict[str, Response | Reference] | None,
|
||||
openapi: OpenAPI,
|
||||
|
|
@ -358,194 +243,688 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute
|
|||
"""
|
||||
Parses an OpenAPI schema dictionary into a list of HTTPRoute objects
|
||||
using the openapi-pydantic library.
|
||||
|
||||
Supports both OpenAPI 3.0.x and 3.1.x versions.
|
||||
"""
|
||||
routes: list[HTTPRoute] = []
|
||||
# Check OpenAPI version to use appropriate model
|
||||
openapi_version = openapi_dict.get("openapi", "")
|
||||
|
||||
try:
|
||||
openapi: OpenAPI = OpenAPI.model_validate(openapi_dict)
|
||||
logger.info(f"Successfully parsed OpenAPI schema version: {openapi.openapi}")
|
||||
if openapi_version.startswith("3.0"):
|
||||
# Use OpenAPI 3.0 models
|
||||
openapi_30 = OpenAPI_30.model_validate(openapi_dict)
|
||||
logger.info(
|
||||
f"Successfully parsed OpenAPI 3.0 schema version: {openapi_30.openapi}"
|
||||
)
|
||||
parser = OpenAPI30Parser(openapi_30)
|
||||
return parser.parse()
|
||||
else:
|
||||
# Default to OpenAPI 3.1 models
|
||||
openapi_31 = OpenAPI.model_validate(openapi_dict)
|
||||
logger.info(
|
||||
f"Successfully parsed OpenAPI 3.1 schema version: {openapi_31.openapi}"
|
||||
)
|
||||
parser = OpenAPI31Parser(openapi_31)
|
||||
return parser.parse()
|
||||
except ValidationError as e:
|
||||
logger.error(f"OpenAPI schema validation failed: {e}")
|
||||
error_details = e.errors()
|
||||
logger.error(f"Validation errors: {error_details}")
|
||||
raise ValueError(f"Invalid OpenAPI schema: {error_details}") from e
|
||||
|
||||
if not openapi.paths:
|
||||
logger.warning("OpenAPI schema has no paths defined.")
|
||||
return []
|
||||
|
||||
for path_str, path_item_obj in openapi.paths.items():
|
||||
if not isinstance(path_item_obj, PathItem):
|
||||
# Base parser class for shared functionality
|
||||
class BaseOpenAPIParser:
|
||||
"""Base class for OpenAPI parsers with common functionality."""
|
||||
|
||||
def _convert_to_parameter_location(self, param_in: str) -> ParameterLocation:
|
||||
"""Convert string parameter location to our ParameterLocation type."""
|
||||
if param_in == "path":
|
||||
return "path"
|
||||
elif param_in == "query":
|
||||
return "query"
|
||||
elif param_in == "header":
|
||||
return "header"
|
||||
elif param_in == "cookie":
|
||||
return "cookie"
|
||||
else:
|
||||
logger.warning(
|
||||
f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})"
|
||||
f"Unknown parameter location: {param_in}, defaulting to 'query'"
|
||||
)
|
||||
continue
|
||||
return "query"
|
||||
|
||||
path_level_params = path_item_obj.parameters
|
||||
|
||||
# Iterate through possible HTTP methods defined in the PathItem model fields
|
||||
# Use model_fields from the class, not the instance
|
||||
for method_lower in PathItem.model_fields.keys():
|
||||
if method_lower not in [
|
||||
"get",
|
||||
"put",
|
||||
"post",
|
||||
"delete",
|
||||
"options",
|
||||
"head",
|
||||
"patch",
|
||||
"trace",
|
||||
]:
|
||||
class OpenAPI31Parser(BaseOpenAPIParser):
|
||||
"""Parser for OpenAPI 3.1 schemas."""
|
||||
|
||||
def __init__(self, openapi: OpenAPI):
|
||||
self.openapi = openapi
|
||||
|
||||
def parse(self) -> list[HTTPRoute]:
|
||||
"""Parse an OpenAPI 3.1 schema into HTTP routes."""
|
||||
routes: list[HTTPRoute] = []
|
||||
|
||||
if not self.openapi.paths:
|
||||
logger.warning("OpenAPI schema has no paths defined.")
|
||||
return []
|
||||
|
||||
for path_str, path_item_obj in self.openapi.paths.items():
|
||||
if not isinstance(path_item_obj, PathItem):
|
||||
logger.warning(
|
||||
f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})"
|
||||
)
|
||||
continue
|
||||
|
||||
operation: Operation | None = getattr(path_item_obj, method_lower, None)
|
||||
path_level_params = path_item_obj.parameters
|
||||
|
||||
if operation and isinstance(operation, Operation):
|
||||
method_upper = cast(HttpMethod, method_lower.upper())
|
||||
logger.debug(f"Processing operation: {method_upper} {path_str}")
|
||||
try:
|
||||
parameters = _extract_parameters(
|
||||
operation.parameters, path_level_params, openapi
|
||||
# Iterate through possible HTTP methods defined in the PathItem model fields
|
||||
# Use model_fields from the class, not the instance
|
||||
for method_lower in PathItem.model_fields.keys():
|
||||
if method_lower not in [
|
||||
"get",
|
||||
"put",
|
||||
"post",
|
||||
"delete",
|
||||
"options",
|
||||
"head",
|
||||
"patch",
|
||||
"trace",
|
||||
]:
|
||||
continue
|
||||
|
||||
operation: Operation | None = getattr(path_item_obj, method_lower, None)
|
||||
|
||||
if operation and isinstance(operation, Operation):
|
||||
method_upper = cast(HttpMethod, method_lower.upper())
|
||||
logger.debug(f"Processing operation: {method_upper} {path_str}")
|
||||
try:
|
||||
parameters = self._extract_parameters(
|
||||
operation.parameters, path_level_params
|
||||
)
|
||||
request_body_info = self._extract_request_body(
|
||||
operation.requestBody
|
||||
)
|
||||
responses = self._extract_responses(operation.responses)
|
||||
|
||||
route = HTTPRoute(
|
||||
path=path_str,
|
||||
method=method_upper,
|
||||
operation_id=operation.operationId,
|
||||
summary=operation.summary,
|
||||
description=operation.description,
|
||||
tags=operation.tags or [],
|
||||
parameters=parameters,
|
||||
request_body=request_body_info,
|
||||
responses=responses,
|
||||
)
|
||||
routes.append(route)
|
||||
logger.info(
|
||||
f"Successfully extracted route: {method_upper} {path_str}"
|
||||
)
|
||||
except Exception as op_error:
|
||||
op_id = operation.operationId or "unknown"
|
||||
logger.error(
|
||||
f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.")
|
||||
return routes
|
||||
|
||||
def _resolve_ref(
|
||||
self, item: Reference | Schema | Parameter | RequestBody | Any
|
||||
) -> Any:
|
||||
"""Resolves a potential Reference object to its target definition."""
|
||||
if isinstance(item, Reference):
|
||||
ref_str = item.ref
|
||||
try:
|
||||
if not ref_str.startswith("#/"):
|
||||
raise ValueError(
|
||||
f"External or non-local reference not supported: {ref_str}"
|
||||
)
|
||||
request_body_info = _extract_request_body(
|
||||
operation.requestBody, openapi
|
||||
parts = ref_str.strip("#/").split("/")
|
||||
target = self.openapi
|
||||
for part in parts:
|
||||
if part.isdigit() and isinstance(target, list):
|
||||
target = target[int(part)]
|
||||
elif isinstance(target, BaseModel):
|
||||
# Use model_extra for fields not explicitly defined (like components types)
|
||||
# Check class fields first, then model_extra
|
||||
if part in target.__class__.model_fields:
|
||||
target = getattr(target, part, None)
|
||||
elif target.model_extra and part in target.model_extra:
|
||||
target = target.model_extra[part]
|
||||
else:
|
||||
# Special handling for components sub-types common structure
|
||||
if part == "components" and hasattr(target, "components"):
|
||||
target = getattr(target, "components")
|
||||
elif hasattr(target, part): # Fallback check
|
||||
target = getattr(target, part, None)
|
||||
else:
|
||||
target = None # Part not found
|
||||
elif isinstance(target, dict):
|
||||
target = target.get(part)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}"
|
||||
)
|
||||
if target is None:
|
||||
raise ValueError(
|
||||
f"Reference part '{part}' not found in path '{ref_str}'"
|
||||
)
|
||||
if isinstance(target, Reference):
|
||||
return self._resolve_ref(target)
|
||||
return target
|
||||
except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e:
|
||||
raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e
|
||||
return item
|
||||
|
||||
def _extract_schema_as_dict(self, schema_obj: Schema | Reference) -> JsonSchema:
|
||||
"""Resolves a schema/reference and returns it as a dictionary."""
|
||||
resolved_schema = self._resolve_ref(schema_obj)
|
||||
if isinstance(resolved_schema, Schema):
|
||||
# Using exclude_none=True might be better than exclude_unset sometimes
|
||||
return resolved_schema.model_dump(
|
||||
mode="json", by_alias=True, exclude_none=True
|
||||
)
|
||||
elif isinstance(resolved_schema, dict):
|
||||
logger.warning(
|
||||
"Resolved schema reference resulted in a dict, not a Schema model."
|
||||
)
|
||||
return resolved_schema
|
||||
else:
|
||||
ref_str = getattr(schema_obj, "ref", "unknown")
|
||||
logger.warning(
|
||||
f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict."
|
||||
)
|
||||
return {}
|
||||
|
||||
def _extract_parameters(
|
||||
self,
|
||||
operation_params: list[Parameter | Reference] | None,
|
||||
path_item_params: list[Parameter | Reference] | None,
|
||||
) -> list[ParameterInfo]:
|
||||
"""Extracts and resolves parameters using corrected attribute names."""
|
||||
extracted_params: list[ParameterInfo] = []
|
||||
seen_params: dict[
|
||||
tuple[str, str], bool
|
||||
] = {} # Use string keys to avoid type issues
|
||||
all_params_refs = (operation_params or []) + (path_item_params or [])
|
||||
|
||||
for param_or_ref in all_params_refs:
|
||||
try:
|
||||
parameter = cast(Parameter, self._resolve_ref(param_or_ref))
|
||||
if not isinstance(parameter, Parameter):
|
||||
# ... (error logging remains the same)
|
||||
continue
|
||||
|
||||
# --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
|
||||
param_in = parameter.param_in # CORRECTED: Use 'param_in'
|
||||
param_location = self._convert_to_parameter_location(param_in)
|
||||
param_schema_obj = (
|
||||
parameter.param_schema
|
||||
) # CORRECTED: Use 'param_schema'
|
||||
# --- *** ---
|
||||
|
||||
param_key = (parameter.name, param_in)
|
||||
if param_key in seen_params:
|
||||
continue
|
||||
seen_params[param_key] = True
|
||||
|
||||
param_schema_dict = {}
|
||||
if param_schema_obj: # Check if schema exists
|
||||
param_schema_dict = self._extract_schema_as_dict(param_schema_obj)
|
||||
elif parameter.content:
|
||||
# Handle complex parameters with 'content'
|
||||
first_media_type = next(iter(parameter.content.values()), None)
|
||||
if (
|
||||
first_media_type and first_media_type.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
param_schema_dict = self._extract_schema_as_dict(
|
||||
first_media_type.media_type_schema
|
||||
)
|
||||
logger.debug(
|
||||
f"Parameter '{parameter.name}' using schema from 'content' field."
|
||||
)
|
||||
|
||||
# Manually create ParameterInfo instance using correct field names
|
||||
param_info = ParameterInfo(
|
||||
name=parameter.name,
|
||||
location=param_location, # Use converted parameter location
|
||||
required=parameter.required,
|
||||
schema=param_schema_dict, # Populate 'schema' field in IR
|
||||
description=parameter.description,
|
||||
)
|
||||
extracted_params.append(param_info)
|
||||
|
||||
except (
|
||||
ValidationError,
|
||||
ValueError,
|
||||
AttributeError,
|
||||
TypeError,
|
||||
) as e: # Added TypeError
|
||||
param_name = getattr(
|
||||
param_or_ref, "name", getattr(param_or_ref, "ref", "unknown")
|
||||
)
|
||||
logger.error(
|
||||
f"Failed to extract parameter '{param_name}': {e}", exc_info=False
|
||||
)
|
||||
|
||||
return extracted_params
|
||||
|
||||
def _extract_request_body(
|
||||
self, request_body_or_ref: RequestBody | Reference | None
|
||||
) -> RequestBodyInfo | None:
|
||||
"""Extracts and resolves the request body using corrected attribute names."""
|
||||
if not request_body_or_ref:
|
||||
return None
|
||||
try:
|
||||
request_body = cast(RequestBody, self._resolve_ref(request_body_or_ref))
|
||||
if not isinstance(request_body, RequestBody):
|
||||
# ... (error logging remains the same)
|
||||
return None
|
||||
|
||||
content_schemas: dict[str, JsonSchema] = {}
|
||||
if request_body.content:
|
||||
for media_type_str, media_type_obj in request_body.content.items():
|
||||
# --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
|
||||
if (
|
||||
isinstance(media_type_obj, MediaType)
|
||||
and media_type_obj.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
# --- *** ---
|
||||
try:
|
||||
# Use the corrected attribute here as well
|
||||
schema_dict = self._extract_schema_as_dict(
|
||||
media_type_obj.media_type_schema
|
||||
)
|
||||
content_schemas[media_type_str] = schema_dict
|
||||
except ValueError as schema_err:
|
||||
logger.error(
|
||||
f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}"
|
||||
)
|
||||
elif not isinstance(media_type_obj, MediaType):
|
||||
logger.warning(
|
||||
f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body."
|
||||
)
|
||||
elif not media_type_obj.media_type_schema: # Corrected check
|
||||
logger.warning(
|
||||
f"Skipping media type '{media_type_str}' in request body because it lacks a schema."
|
||||
)
|
||||
|
||||
return RequestBodyInfo(
|
||||
required=request_body.required,
|
||||
content_schema=content_schemas,
|
||||
description=request_body.description,
|
||||
)
|
||||
except (ValidationError, ValueError, AttributeError) as e:
|
||||
ref_name = getattr(request_body_or_ref, "ref", "unknown")
|
||||
logger.error(
|
||||
f"Failed to extract request body '{ref_name}': {e}", exc_info=False
|
||||
)
|
||||
return None
|
||||
|
||||
def _extract_responses(
|
||||
self,
|
||||
operation_responses: dict[str, Response | Reference] | None,
|
||||
) -> dict[str, ResponseInfo]:
|
||||
"""Extracts and resolves response information for an operation."""
|
||||
extracted_responses: dict[str, ResponseInfo] = {}
|
||||
if not operation_responses:
|
||||
return extracted_responses
|
||||
|
||||
for status_code, resp_or_ref in operation_responses.items():
|
||||
try:
|
||||
response = cast(Response, self._resolve_ref(resp_or_ref))
|
||||
if not isinstance(response, Response):
|
||||
ref_str = getattr(resp_or_ref, "ref", "unknown")
|
||||
logger.warning(
|
||||
f"Expected Response after resolving ref '{ref_str}' for status code {status_code}, got {type(response)}. Skipping."
|
||||
)
|
||||
responses = _extract_responses(operation.responses, openapi)
|
||||
continue
|
||||
|
||||
route = HTTPRoute(
|
||||
path=path_str,
|
||||
method=method_upper,
|
||||
operation_id=operation.operationId,
|
||||
summary=operation.summary,
|
||||
description=operation.description,
|
||||
tags=operation.tags or [],
|
||||
parameters=parameters,
|
||||
request_body=request_body_info,
|
||||
responses=responses,
|
||||
content_schemas: dict[str, JsonSchema] = {}
|
||||
if response.content:
|
||||
for media_type_str, media_type_obj in response.content.items():
|
||||
if (
|
||||
isinstance(media_type_obj, MediaType)
|
||||
and media_type_obj.media_type_schema
|
||||
):
|
||||
try:
|
||||
schema_dict = self._extract_schema_as_dict(
|
||||
media_type_obj.media_type_schema
|
||||
)
|
||||
content_schemas[media_type_str] = schema_dict
|
||||
except ValueError as schema_err:
|
||||
logger.error(
|
||||
f"Failed to extract schema for media type '{media_type_str}' in response {status_code}: {schema_err}"
|
||||
)
|
||||
|
||||
resp_info = ResponseInfo(
|
||||
description=response.description, content_schema=content_schemas
|
||||
)
|
||||
extracted_responses[str(status_code)] = resp_info
|
||||
|
||||
except (ValidationError, ValueError, AttributeError) as e:
|
||||
ref_name = getattr(resp_or_ref, "ref", "unknown")
|
||||
logger.error(
|
||||
f"Failed to extract response for status code {status_code} "
|
||||
f"from reference '{ref_name}': {e}",
|
||||
exc_info=False,
|
||||
)
|
||||
|
||||
return extracted_responses
|
||||
|
||||
|
||||
class OpenAPI30Parser(BaseOpenAPIParser):
|
||||
"""Parser for OpenAPI 3.0 schemas."""
|
||||
|
||||
def __init__(self, openapi: OpenAPI_30):
|
||||
self.openapi = openapi
|
||||
|
||||
def parse(self) -> list[HTTPRoute]:
|
||||
"""Parse an OpenAPI 3.0 schema into HTTP routes."""
|
||||
routes: list[HTTPRoute] = []
|
||||
|
||||
if not self.openapi.paths:
|
||||
logger.warning("OpenAPI schema has no paths defined.")
|
||||
return []
|
||||
|
||||
for path_str, path_item_obj in self.openapi.paths.items():
|
||||
if not isinstance(path_item_obj, PathItem_30):
|
||||
logger.warning(
|
||||
f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})"
|
||||
)
|
||||
continue
|
||||
|
||||
path_level_params = path_item_obj.parameters
|
||||
|
||||
# Iterate through possible HTTP methods defined in the PathItem model fields
|
||||
# Use model_fields from the class, not the instance
|
||||
for method_lower in PathItem_30.model_fields.keys():
|
||||
if method_lower not in [
|
||||
"get",
|
||||
"put",
|
||||
"post",
|
||||
"delete",
|
||||
"options",
|
||||
"head",
|
||||
"patch",
|
||||
"trace",
|
||||
]:
|
||||
continue
|
||||
|
||||
operation: Operation_30 | None = getattr(
|
||||
path_item_obj, method_lower, None
|
||||
)
|
||||
|
||||
if operation and isinstance(operation, Operation_30):
|
||||
method_upper = cast(HttpMethod, method_lower.upper())
|
||||
logger.debug(f"Processing operation: {method_upper} {path_str}")
|
||||
try:
|
||||
parameters = self._extract_parameters(
|
||||
operation.parameters, path_level_params
|
||||
)
|
||||
request_body_info = self._extract_request_body(
|
||||
operation.requestBody
|
||||
)
|
||||
responses = self._extract_responses(operation.responses)
|
||||
|
||||
route = HTTPRoute(
|
||||
path=path_str,
|
||||
method=method_upper,
|
||||
operation_id=operation.operationId,
|
||||
summary=operation.summary,
|
||||
description=operation.description,
|
||||
tags=operation.tags or [],
|
||||
parameters=parameters,
|
||||
request_body=request_body_info,
|
||||
responses=responses,
|
||||
)
|
||||
routes.append(route)
|
||||
logger.info(
|
||||
f"Successfully extracted route: {method_upper} {path_str}"
|
||||
)
|
||||
except Exception as op_error:
|
||||
op_id = operation.operationId or "unknown"
|
||||
logger.error(
|
||||
f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.")
|
||||
return routes
|
||||
|
||||
def _resolve_ref(
|
||||
self, item: Reference_30 | Schema_30 | Parameter_30 | RequestBody_30 | Any
|
||||
) -> Any:
|
||||
"""Resolves a potential Reference object to its target definition for OpenAPI 3.0."""
|
||||
if isinstance(item, Reference_30):
|
||||
ref_str = item.ref
|
||||
try:
|
||||
if not ref_str.startswith("#/"):
|
||||
raise ValueError(
|
||||
f"External or non-local reference not supported: {ref_str}"
|
||||
)
|
||||
routes.append(route)
|
||||
logger.info(
|
||||
f"Successfully extracted route: {method_upper} {path_str}"
|
||||
parts = ref_str.strip("#/").split("/")
|
||||
target = self.openapi
|
||||
for part in parts:
|
||||
if part.isdigit() and isinstance(target, list):
|
||||
target = target[int(part)]
|
||||
elif isinstance(target, BaseModel):
|
||||
# Use model_extra for fields not explicitly defined (like components types)
|
||||
# Check class fields first, then model_extra
|
||||
if part in target.__class__.model_fields:
|
||||
target = getattr(target, part, None)
|
||||
elif target.model_extra and part in target.model_extra:
|
||||
target = target.model_extra[part]
|
||||
else:
|
||||
# Special handling for components sub-types common structure
|
||||
if part == "components" and hasattr(target, "components"):
|
||||
target = getattr(target, "components")
|
||||
elif hasattr(target, part): # Fallback check
|
||||
target = getattr(target, part, None)
|
||||
else:
|
||||
target = None # Part not found
|
||||
elif isinstance(target, dict):
|
||||
target = target.get(part)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}"
|
||||
)
|
||||
if target is None:
|
||||
raise ValueError(
|
||||
f"Reference part '{part}' not found in path '{ref_str}'"
|
||||
)
|
||||
if isinstance(target, Reference_30):
|
||||
return self._resolve_ref(target)
|
||||
return target
|
||||
except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e:
|
||||
raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e
|
||||
return item
|
||||
|
||||
def _extract_schema_as_dict(
|
||||
self, schema_obj: Schema_30 | Reference_30
|
||||
) -> JsonSchema:
|
||||
"""Resolves a schema/reference and returns it as a dictionary for OpenAPI 3.0."""
|
||||
resolved_schema = self._resolve_ref(schema_obj)
|
||||
if isinstance(resolved_schema, Schema_30):
|
||||
# Using exclude_none=True might be better than exclude_unset sometimes
|
||||
return resolved_schema.model_dump(
|
||||
mode="json", by_alias=True, exclude_none=True
|
||||
)
|
||||
elif isinstance(resolved_schema, dict):
|
||||
logger.warning(
|
||||
"Resolved schema reference resulted in a dict, not a Schema model."
|
||||
)
|
||||
return resolved_schema
|
||||
else:
|
||||
ref_str = getattr(schema_obj, "ref", "unknown")
|
||||
logger.warning(
|
||||
f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict."
|
||||
)
|
||||
return {}
|
||||
|
||||
def _extract_parameters(
|
||||
self,
|
||||
operation_params: list[Parameter_30 | Reference_30] | None,
|
||||
path_item_params: list[Parameter_30 | Reference_30] | None,
|
||||
) -> list[ParameterInfo]:
|
||||
"""Extracts and resolves parameters for OpenAPI 3.0."""
|
||||
extracted_params: list[ParameterInfo] = []
|
||||
seen_params: dict[
|
||||
tuple[str, str], bool
|
||||
] = {} # Use string keys to avoid type issues
|
||||
all_params_refs = (operation_params or []) + (path_item_params or [])
|
||||
|
||||
for param_or_ref in all_params_refs:
|
||||
try:
|
||||
parameter = cast(Parameter_30, self._resolve_ref(param_or_ref))
|
||||
if not isinstance(parameter, Parameter_30):
|
||||
logger.warning(
|
||||
f"Expected Parameter after resolving reference, got {type(parameter)}. Skipping."
|
||||
)
|
||||
except Exception as op_error:
|
||||
op_id = operation.operationId or "unknown"
|
||||
logger.error(
|
||||
f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
|
||||
exc_info=True,
|
||||
continue
|
||||
|
||||
# OpenAPI 3.0 uses 'in' field for parameter location
|
||||
param_in = parameter.param_in
|
||||
param_location = self._convert_to_parameter_location(param_in)
|
||||
param_schema_obj = parameter.param_schema
|
||||
|
||||
param_key = (parameter.name, param_in)
|
||||
if param_key in seen_params:
|
||||
continue
|
||||
seen_params[param_key] = True
|
||||
|
||||
param_schema_dict = {}
|
||||
if param_schema_obj: # Check if schema exists
|
||||
param_schema_dict = self._extract_schema_as_dict(param_schema_obj)
|
||||
elif parameter.content:
|
||||
# Handle complex parameters with 'content'
|
||||
first_media_type = next(iter(parameter.content.values()), None)
|
||||
if first_media_type and first_media_type.media_type_schema:
|
||||
param_schema_dict = self._extract_schema_as_dict(
|
||||
first_media_type.media_type_schema
|
||||
)
|
||||
logger.debug(
|
||||
f"Parameter '{parameter.name}' using schema from 'content' field."
|
||||
)
|
||||
|
||||
# Manually create ParameterInfo instance using correct field names
|
||||
param_info = ParameterInfo(
|
||||
name=parameter.name,
|
||||
location=param_location, # Use converted parameter location
|
||||
required=parameter.required,
|
||||
schema=param_schema_dict, # Populate 'schema' field in IR
|
||||
description=parameter.description,
|
||||
)
|
||||
extracted_params.append(param_info)
|
||||
|
||||
except (
|
||||
ValidationError,
|
||||
ValueError,
|
||||
AttributeError,
|
||||
TypeError,
|
||||
) as e: # Added TypeError
|
||||
param_name = getattr(
|
||||
param_or_ref, "name", getattr(param_or_ref, "ref", "unknown")
|
||||
)
|
||||
logger.error(
|
||||
f"Failed to extract parameter '{param_name}': {e}", exc_info=False
|
||||
)
|
||||
|
||||
return extracted_params
|
||||
|
||||
def _extract_request_body(
|
||||
self, request_body_or_ref: RequestBody_30 | Reference_30 | None
|
||||
) -> RequestBodyInfo | None:
|
||||
"""Extracts request body information for OpenAPI 3.0 using correct attribute names."""
|
||||
if request_body_or_ref is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
request_body = cast(RequestBody_30, self._resolve_ref(request_body_or_ref))
|
||||
|
||||
if not isinstance(request_body, RequestBody_30):
|
||||
logger.warning(
|
||||
f"Expected RequestBody after resolving reference, got {type(request_body)}. Returning None."
|
||||
)
|
||||
return None
|
||||
|
||||
request_body_info = RequestBodyInfo(
|
||||
required=request_body.required,
|
||||
description=request_body.description,
|
||||
)
|
||||
|
||||
# Process content field for request body schemas
|
||||
if request_body.content:
|
||||
for media_type_key, media_type_obj in request_body.content.items():
|
||||
if (
|
||||
media_type_obj and media_type_obj.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
schema_dict = self._extract_schema_as_dict(
|
||||
media_type_obj.media_type_schema
|
||||
)
|
||||
request_body_info.content_schema[media_type_key] = schema_dict
|
||||
|
||||
return request_body_info
|
||||
|
||||
except (ValidationError, ValueError, AttributeError) as e:
|
||||
ref_str = getattr(request_body_or_ref, "ref", "unknown")
|
||||
logger.error(
|
||||
f"Failed to extract request body info from reference '{ref_str}': {e}",
|
||||
exc_info=False,
|
||||
)
|
||||
return None
|
||||
|
||||
def _extract_responses(
|
||||
self,
|
||||
operation_responses: dict[str, Response_30 | Reference_30] | None,
|
||||
) -> dict[str, ResponseInfo]:
|
||||
"""Extracts response information from an OpenAPI 3.0 operation's responses."""
|
||||
extracted_responses: dict[str, ResponseInfo] = {}
|
||||
if not operation_responses:
|
||||
return extracted_responses
|
||||
|
||||
for status_code, response_or_ref in operation_responses.items():
|
||||
try:
|
||||
# Skip 'default' response for simplicity if needed
|
||||
# if status_code == "default":
|
||||
# continue
|
||||
|
||||
response = cast(Response_30, self._resolve_ref(response_or_ref))
|
||||
|
||||
if not isinstance(response, Response_30):
|
||||
logger.warning(
|
||||
f"Expected Response after resolving reference for status code {status_code}, "
|
||||
f"got {type(response)}. Skipping."
|
||||
)
|
||||
continue
|
||||
|
||||
logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.")
|
||||
return routes
|
||||
response_info = ResponseInfo(description=response.description)
|
||||
|
||||
# Extract content schemas if present
|
||||
if response.content:
|
||||
for media_type_key, media_type_obj in response.content.items():
|
||||
if (
|
||||
media_type_obj and media_type_obj.media_type_schema
|
||||
): # CORRECTED: Use 'media_type_schema'
|
||||
schema_dict = self._extract_schema_as_dict(
|
||||
media_type_obj.media_type_schema
|
||||
)
|
||||
response_info.content_schema[media_type_key] = schema_dict
|
||||
|
||||
# --- Example Usage (Optional) ---
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
extracted_responses[status_code] = response_info
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s"
|
||||
) # Set to INFO
|
||||
except (ValidationError, ValueError, AttributeError) as e:
|
||||
ref_str = getattr(response_or_ref, "ref", "unknown")
|
||||
logger.error(
|
||||
f"Failed to extract response info for status code {status_code} "
|
||||
f"from reference '{ref_str}': {e}",
|
||||
exc_info=False,
|
||||
)
|
||||
|
||||
petstore_schema = {
|
||||
"openapi": "3.1.0", # Keep corrected version
|
||||
"info": {"title": "Simple Pet Store API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/pets": {
|
||||
"get": {
|
||||
"summary": "list all pets",
|
||||
"operationId": "listPets",
|
||||
"tags": ["pets"],
|
||||
"parameters": [
|
||||
{
|
||||
"name": "limit",
|
||||
"in": "query",
|
||||
"description": "How many items to return",
|
||||
"required": False,
|
||||
"schema": {"type": "integer", "format": "int32"},
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "A paged array of pets"}},
|
||||
},
|
||||
"post": {
|
||||
"summary": "Create a pet",
|
||||
"operationId": "createPet",
|
||||
"tags": ["pets"],
|
||||
"requestBody": {"$ref": "#/components/requestBodies/PetBody"},
|
||||
"responses": {"201": {"description": "Null response"}},
|
||||
},
|
||||
},
|
||||
"/pets/{petId}": {
|
||||
"get": {
|
||||
"summary": "Info for a specific pet",
|
||||
"operationId": "showPetById",
|
||||
"tags": ["pets"],
|
||||
"parameters": [
|
||||
{
|
||||
"name": "petId",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"description": "The id of the pet",
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "X-Request-ID",
|
||||
"in": "header",
|
||||
"required": False,
|
||||
"schema": {"type": "string", "format": "uuid"},
|
||||
},
|
||||
],
|
||||
"responses": {"200": {"description": "Information about the pet"}},
|
||||
},
|
||||
"parameters": [ # Path level parameter example
|
||||
{
|
||||
"name": "traceId",
|
||||
"in": "header",
|
||||
"description": "Common trace ID",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Pet": {
|
||||
"type": "object",
|
||||
"required": ["id", "name"],
|
||||
"properties": {
|
||||
"id": {"type": "integer", "format": "int64"},
|
||||
"name": {"type": "string"},
|
||||
"tag": {"type": "string"},
|
||||
},
|
||||
}
|
||||
},
|
||||
"requestBodies": {
|
||||
"PetBody": {
|
||||
"description": "Pet object",
|
||||
"required": True,
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Pet"}
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
print("--- Parsing Pet Store Schema using openapi-pydantic (Corrected) ---")
|
||||
try:
|
||||
http_routes = parse_openapi_to_http_routes(petstore_schema)
|
||||
print(f"\n--- Extracted {len(http_routes)} Routes ---")
|
||||
for i, route in enumerate(http_routes):
|
||||
print(f"\nRoute {i + 1}:")
|
||||
# Use model_dump for clean JSON-like output, show aliases from IR model
|
||||
print(
|
||||
json.dumps(route.model_dump(by_alias=True, exclude_none=True), indent=2)
|
||||
) # exclude_none is often cleaner
|
||||
except ValueError as e:
|
||||
print(f"\nError parsing schema: {e}")
|
||||
except Exception as e:
|
||||
print(f"\nAn unexpected error occurred: {e}")
|
||||
return extracted_responses
|
||||
|
||||
|
||||
def clean_schema_for_display(schema: JsonSchema | None) -> JsonSchema | None:
|
||||
|
|
|
|||
|
|
@ -311,6 +311,172 @@ def fastapi_route_map(parsed_fastapi_routes):
|
|||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openapi_30_schema() -> dict[str, Any]:
|
||||
"""Fixture that returns a simple OpenAPI 3.0.0 schema."""
|
||||
return {
|
||||
"openapi": "3.0.0",
|
||||
"info": {"title": "Simple API (OpenAPI 3.0)", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"summary": "List all items",
|
||||
"operationId": "listItems",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "limit",
|
||||
"in": "query",
|
||||
"description": "How many items to return",
|
||||
"required": False,
|
||||
"schema": {"type": "integer"},
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "A list of items"}},
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openapi_31_schema() -> dict[str, Any]:
|
||||
"""Fixture that returns a simple OpenAPI 3.1.0 schema."""
|
||||
return {
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": "Simple API (OpenAPI 3.1)", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"summary": "List all items",
|
||||
"operationId": "listItems",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "limit",
|
||||
"in": "query",
|
||||
"description": "How many items to return",
|
||||
"required": False,
|
||||
"schema": {"type": "integer"},
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "A list of items"}},
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openapi_30_with_references() -> dict[str, Any]:
|
||||
"""OpenAPI 3.0 schema with references to test resolution."""
|
||||
return {
|
||||
"openapi": "3.0.0",
|
||||
"info": {"title": "API with References (3.0)", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/products": {
|
||||
"post": {
|
||||
"summary": "Create product",
|
||||
"operationId": "createProduct",
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Product"}
|
||||
}
|
||||
},
|
||||
"required": True,
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "Product created",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Product"}
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Product": {
|
||||
"type": "object",
|
||||
"required": ["name", "price"],
|
||||
"properties": {
|
||||
"id": {"type": "string", "format": "uuid"},
|
||||
"name": {"type": "string"},
|
||||
"price": {"type": "number"},
|
||||
"category": {"$ref": "#/components/schemas/Category"},
|
||||
},
|
||||
},
|
||||
"Category": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "integer"},
|
||||
"name": {"type": "string"},
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openapi_31_with_references() -> dict[str, Any]:
|
||||
"""OpenAPI 3.1 schema with references to test resolution."""
|
||||
return {
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": "API with References (3.1)", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/products": {
|
||||
"post": {
|
||||
"summary": "Create product",
|
||||
"operationId": "createProduct",
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Product"}
|
||||
}
|
||||
},
|
||||
"required": True,
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "Product created",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Product"}
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"Product": {
|
||||
"type": "object",
|
||||
"required": ["name", "price"],
|
||||
"properties": {
|
||||
"id": {"type": "string", "format": "uuid"},
|
||||
"name": {"type": "string"},
|
||||
"price": {"type": "number"},
|
||||
"category": {"$ref": "#/components/schemas/Category"},
|
||||
},
|
||||
},
|
||||
"Category": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {"type": "integer"},
|
||||
"name": {"type": "string"},
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# --- Tests for PetStore schema --- #
|
||||
|
||||
|
||||
|
|
@ -764,3 +930,166 @@ def test_fastapi_post_query_parameter_names(fastapi_route_map):
|
|||
param_names = [p.name for p in query_params]
|
||||
assert "file_name" in param_names
|
||||
assert "content_type" in param_names
|
||||
|
||||
|
||||
def test_openapi_30_compatibility(openapi_30_schema):
|
||||
"""Test that OpenAPI 3.0 schemas can be parsed correctly."""
|
||||
# This will raise an exception if the parser doesn't support 3.0.0
|
||||
routes = parse_openapi_to_http_routes(openapi_30_schema)
|
||||
|
||||
# Verify the route was parsed correctly
|
||||
assert len(routes) == 1
|
||||
route = routes[0]
|
||||
assert route.method == "GET"
|
||||
assert route.path == "/items"
|
||||
assert route.operation_id == "listItems"
|
||||
assert len(route.parameters) == 1
|
||||
assert route.parameters[0].name == "limit"
|
||||
|
||||
|
||||
def test_openapi_31_compatibility(openapi_31_schema):
|
||||
"""Test that OpenAPI 3.1 schemas can be parsed correctly."""
|
||||
routes = parse_openapi_to_http_routes(openapi_31_schema)
|
||||
|
||||
# Verify the route was parsed correctly
|
||||
assert len(routes) == 1
|
||||
route = routes[0]
|
||||
assert route.method == "GET"
|
||||
assert route.path == "/items"
|
||||
assert route.operation_id == "listItems"
|
||||
assert len(route.parameters) == 1
|
||||
assert route.parameters[0].name == "limit"
|
||||
|
||||
|
||||
def test_version_detection_logic():
|
||||
"""Test that the version detection logic correctly identifies 3.0 vs 3.1 schemas."""
|
||||
# Test 3.0 variations
|
||||
for version in ["3.0.0", "3.0.1", "3.0.3"]:
|
||||
schema = {
|
||||
"openapi": version,
|
||||
"info": {"title": "Test", "version": "1.0.0"},
|
||||
"paths": {},
|
||||
}
|
||||
try:
|
||||
parse_openapi_to_http_routes(schema)
|
||||
# Expect no error
|
||||
except Exception as e:
|
||||
pytest.fail(f"Failed to parse OpenAPI {version} schema: {e}")
|
||||
|
||||
# Test 3.1 variations
|
||||
for version in ["3.1.0", "3.1.1"]:
|
||||
schema = {
|
||||
"openapi": version,
|
||||
"info": {"title": "Test", "version": "1.0.0"},
|
||||
"paths": {},
|
||||
}
|
||||
try:
|
||||
parse_openapi_to_http_routes(schema)
|
||||
# Expect no error
|
||||
except Exception as e:
|
||||
pytest.fail(f"Failed to parse OpenAPI {version} schema: {e}")
|
||||
|
||||
|
||||
def test_openapi_30_reference_resolution(openapi_30_with_references):
|
||||
"""Test that references are correctly resolved in OpenAPI 3.0 schemas."""
|
||||
routes = parse_openapi_to_http_routes(openapi_30_with_references)
|
||||
|
||||
assert len(routes) == 1
|
||||
route = routes[0]
|
||||
assert route.method == "POST"
|
||||
assert route.path == "/products"
|
||||
|
||||
# Check request body
|
||||
assert route.request_body is not None
|
||||
assert route.request_body.required is True
|
||||
assert "application/json" in route.request_body.content_schema
|
||||
|
||||
# Check schema structure
|
||||
json_schema = route.request_body.content_schema["application/json"]
|
||||
assert json_schema["type"] == "object"
|
||||
assert "properties" in json_schema
|
||||
assert set(json_schema["required"]) == {"name", "price"}
|
||||
|
||||
# Check primary fields are properly resolved
|
||||
props = json_schema["properties"]
|
||||
assert "id" in props
|
||||
assert "name" in props
|
||||
assert "price" in props
|
||||
assert "category" in props
|
||||
|
||||
# The category might be a reference or resolved object
|
||||
category = props["category"]
|
||||
# Either it's directly resolved with properties
|
||||
# or it still has a $ref field
|
||||
assert "properties" in category or "$ref" in category
|
||||
|
||||
|
||||
def test_openapi_31_reference_resolution(openapi_31_with_references):
|
||||
"""Test that references are correctly resolved in OpenAPI 3.1 schemas."""
|
||||
routes = parse_openapi_to_http_routes(openapi_31_with_references)
|
||||
|
||||
assert len(routes) == 1
|
||||
route = routes[0]
|
||||
assert route.method == "POST"
|
||||
assert route.path == "/products"
|
||||
|
||||
# Check request body
|
||||
assert route.request_body is not None
|
||||
assert route.request_body.required is True
|
||||
assert "application/json" in route.request_body.content_schema
|
||||
|
||||
# Check schema structure
|
||||
json_schema = route.request_body.content_schema["application/json"]
|
||||
assert json_schema["type"] == "object"
|
||||
assert "properties" in json_schema
|
||||
assert set(json_schema["required"]) == {"name", "price"}
|
||||
|
||||
# Check primary fields are properly resolved
|
||||
props = json_schema["properties"]
|
||||
assert "id" in props
|
||||
assert "name" in props
|
||||
assert "price" in props
|
||||
assert "category" in props
|
||||
|
||||
# The category might be a reference or resolved object
|
||||
category = props["category"]
|
||||
# Either it's directly resolved with properties
|
||||
# or it still has a $ref field
|
||||
assert "properties" in category or "$ref" in category
|
||||
|
||||
|
||||
def test_consistent_output_across_versions(
|
||||
openapi_30_with_references, openapi_31_with_references
|
||||
):
|
||||
"""Test that both parsers produce equivalent output for equivalent schemas."""
|
||||
routes_30 = parse_openapi_to_http_routes(openapi_30_with_references)
|
||||
routes_31 = parse_openapi_to_http_routes(openapi_31_with_references)
|
||||
|
||||
# Convert to dict for easier comparison
|
||||
route_30_dict = routes_30[0].model_dump(exclude_none=True)
|
||||
route_31_dict = routes_31[0].model_dump(exclude_none=True)
|
||||
|
||||
# They should be identical except for version-specific differences
|
||||
# Compare path
|
||||
assert route_30_dict["path"] == route_31_dict["path"]
|
||||
# Compare method
|
||||
assert route_30_dict["method"] == route_31_dict["method"]
|
||||
# Compare operation_id
|
||||
assert route_30_dict["operation_id"] == route_31_dict["operation_id"]
|
||||
# Compare parameters
|
||||
assert len(route_30_dict["parameters"]) == len(route_31_dict["parameters"])
|
||||
# Compare request body
|
||||
assert (
|
||||
route_30_dict["request_body"]["required"]
|
||||
== route_31_dict["request_body"]["required"]
|
||||
)
|
||||
# Compare response structure
|
||||
assert "201" in route_30_dict["responses"] and "201" in route_31_dict["responses"]
|
||||
# The schemas should contain the same essential fields
|
||||
schema_30 = route_30_dict["request_body"]["content_schema"]["application/json"][
|
||||
"properties"
|
||||
]
|
||||
schema_31 = route_31_dict["request_body"]["content_schema"]["application/json"][
|
||||
"properties"
|
||||
]
|
||||
assert set(schema_30.keys()) == set(schema_31.keys())
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue