"""OpenAPI and Swagger specification parser. Extends JsonSchemaParser to handle OpenAPI 2.0 (Swagger), 3.0, and 3.1 specifications, including paths, operations, parameters, and request/response bodies. """ from __future__ import annotations import fnmatch import re from collections import defaultdict from contextlib import nullcontext from enum import Enum from functools import cached_property from pathlib import Path from re import Pattern from typing import TYPE_CHECKING, Any, ClassVar, Optional, TypeVar, Union from warnings import warn from pydantic import Field from typing_extensions import Unpack from datamodel_code_generator import ( Error, OpenAPIScope, YamlValue, load_data, snooper_to_methods, ) from datamodel_code_generator.enums import OpenAPIVersion, VersionMode from datamodel_code_generator.parser.base import get_special_path from datamodel_code_generator.parser.jsonschema import ( JsonSchemaObject, JsonSchemaParser, get_model_by_path, ) from datamodel_code_generator.reference import FieldNameResolver, is_url, snake_to_upper_camel from datamodel_code_generator.types import ( DataType, EmptyDataType, ) from datamodel_code_generator.util import BaseModel if TYPE_CHECKING: from urllib.parse import ParseResult from datamodel_code_generator._types import OpenAPIParserConfigDict from datamodel_code_generator.config import OpenAPIParserConfig from datamodel_code_generator.model import DataModelFieldBase from datamodel_code_generator.parser.schema_version import OpenAPISchemaFeatures RE_APPLICATION_JSON_PATTERN: Pattern[str] = re.compile(r"^application/.*json$") OPERATION_NAMES: list[str] = [ "get", "put", "post", "delete", "patch", "head", "options", "trace", ] class ParameterLocation(Enum): """Represent OpenAPI parameter locations.""" query = "query" header = "header" path = "path" cookie = "cookie" BaseModelT = TypeVar("BaseModelT", bound=BaseModel) class ReferenceObject(BaseModel): """Represent an OpenAPI reference object ($ref).""" ref: str = Field(..., alias="$ref") class ExampleObject(BaseModel): """Represent an OpenAPI example object.""" summary: Optional[str] = None # noqa: UP045 description: Optional[str] = None # noqa: UP045 value: YamlValue = None externalValue: Optional[str] = None # noqa: N815, UP045 class MediaObject(BaseModel): """Represent an OpenAPI media type object.""" schema_: Optional[Union[ReferenceObject, JsonSchemaObject]] = Field(None, alias="schema") # noqa: UP007, UP045 example: YamlValue = None examples: Optional[Union[str, ReferenceObject, ExampleObject]] = None # noqa: UP007, UP045 class ParameterObject(BaseModel): """Represent an OpenAPI parameter object.""" name: Optional[str] = None # noqa: UP045 in_: Optional[ParameterLocation] = Field(None, alias="in") # noqa: UP045 description: Optional[str] = None # noqa: UP045 required: bool = False deprecated: bool = False schema_: Optional[JsonSchemaObject] = Field(None, alias="schema") # noqa: UP045 example: YamlValue = None examples: Optional[Union[str, ReferenceObject, ExampleObject]] = None # noqa: UP007, UP045 content: dict[str, MediaObject] = Field(default_factory=dict) class HeaderObject(BaseModel): """Represent an OpenAPI header object.""" description: Optional[str] = None # noqa: UP045 required: bool = False deprecated: bool = False schema_: Optional[JsonSchemaObject] = Field(None, alias="schema") # noqa: UP045 example: YamlValue = None examples: Optional[Union[str, ReferenceObject, ExampleObject]] = None # noqa: UP007, UP045 content: dict[str, MediaObject] = Field(default_factory=dict) class RequestBodyObject(BaseModel): """Represent an OpenAPI request body object.""" description: Optional[str] = None # noqa: UP045 content: dict[str, MediaObject] = Field(default_factory=dict) required: bool = False class ResponseObject(BaseModel): """Represent an OpenAPI response object.""" description: Optional[str] = None # noqa: UP045 headers: dict[str, ParameterObject] = Field(default_factory=dict) content: dict[Union[str, int], MediaObject] = Field(default_factory=dict) # noqa: UP007 class Operation(BaseModel): """Represent an OpenAPI operation object.""" tags: list[str] = Field(default_factory=list) summary: Optional[str] = None # noqa: UP045 description: Optional[str] = None # noqa: UP045 operationId: Optional[str] = None # noqa: N815, UP045 parameters: list[Union[ReferenceObject, ParameterObject]] = Field(default_factory=list) # noqa: UP007 requestBody: Optional[Union[ReferenceObject, RequestBodyObject]] = None # noqa: N815, UP007, UP045 responses: dict[Union[str, int], Union[ReferenceObject, ResponseObject]] = Field(default_factory=dict) # noqa: UP007 deprecated: bool = False class ComponentsObject(BaseModel): """Represent an OpenAPI components object.""" schemas: dict[str, Union[ReferenceObject, JsonSchemaObject]] = Field(default_factory=dict) # noqa: UP007 responses: dict[str, Union[ReferenceObject, ResponseObject]] = Field(default_factory=dict) # noqa: UP007 examples: dict[str, Union[ReferenceObject, ExampleObject]] = Field(default_factory=dict) # noqa: UP007 requestBodies: dict[str, Union[ReferenceObject, RequestBodyObject]] = Field(default_factory=dict) # noqa: N815, UP007 headers: dict[str, Union[ReferenceObject, HeaderObject]] = Field(default_factory=dict) # noqa: UP007 @snooper_to_methods() class OpenAPIParser(JsonSchemaParser): """Parser for OpenAPI 2.0/3.0/3.1 and Swagger specifications.""" SCHEMA_PATHS: ClassVar[list[str]] = ["#/components/schemas"] @cached_property def schema_features(self) -> OpenAPISchemaFeatures: """Get schema features based on config or detected OpenAPI version.""" from datamodel_code_generator.parser.schema_version import ( # noqa: PLC0415 OpenAPISchemaFeatures, detect_openapi_version, ) config_version = getattr(self.config, "openapi_version", None) if config_version is not None and config_version != OpenAPIVersion.Auto: return OpenAPISchemaFeatures.from_openapi_version(config_version) version = detect_openapi_version(self.raw_obj) if self.raw_obj else OpenAPIVersion.Auto return OpenAPISchemaFeatures.from_openapi_version(version) _config_class_name: ClassVar[str] = "OpenAPIParserConfig" def __init__( self, source: str | Path | list[Path] | ParseResult, *, config: OpenAPIParserConfig | None = None, **options: Unpack[OpenAPIParserConfigDict], ) -> None: """Initialize the OpenAPI parser with extensive configuration options.""" if config is None and options.get("wrap_string_literal") is None: options["wrap_string_literal"] = False super().__init__(source=source, config=config, **options) # type: ignore[arg-type] self.open_api_scopes: list[OpenAPIScope] = self.config.openapi_scopes or [OpenAPIScope.Schemas] # ty: ignore self.include_path_parameters: bool = self.config.include_path_parameters # ty: ignore self.use_status_code_in_response_name: bool = self.config.use_status_code_in_response_name # ty: ignore self.openapi_include_paths: list[str] | None = self.config.openapi_include_paths # ty: ignore if self.openapi_include_paths and OpenAPIScope.Paths not in self.open_api_scopes: warn( "--openapi-include-paths has no effect without --openapi-scopes paths", stacklevel=2, ) self._discriminator_schemas: dict[str, dict[str, Any]] = {} self._discriminator_subtypes: dict[str, list[str]] = defaultdict(list) def get_ref_model(self, ref: str) -> dict[str, Any]: """Resolve a reference to its model definition.""" ref_file, ref_path = self.model_resolver.resolve_ref(ref).split("#", 1) ref_body = self._get_ref_body(ref_file) if ref_file else self.raw_obj return get_model_by_path(ref_body, ref_path.split("/")[1:]) def get_data_type(self, obj: JsonSchemaObject) -> DataType: """Get data type from JSON schema object, handling OpenAPI nullable semantics. Uses schema_features.nullable_keyword to handle version differences: - OpenAPI 3.0: nullable: true is valid, convert to type array when strict_nullable - OpenAPI 3.1: nullable is deprecated, use type: ["string", "null"] instead """ if obj.nullable: if self.schema_features.nullable_keyword: # OpenAPI 3.0: nullable: true is the standard way if self.strict_nullable and isinstance(obj.type, str): obj.type = [obj.type, "null"] else: # OpenAPI 3.1+: nullable is deprecated, still process but warn in Strict mode if self.config.schema_version_mode == VersionMode.Strict: warn( 'nullable keyword is deprecated in OpenAPI 3.1, use type: ["string", "null"] instead', DeprecationWarning, stacklevel=2, ) # Still convert to type array for compatibility if self.strict_nullable and isinstance(obj.type, str): obj.type = [obj.type, "null"] return super().get_data_type(obj) def _normalize_discriminator_mapping_ref(self, mapping_value: str) -> str: # noqa: PLR6301 """Normalize a discriminator mapping value to a full $ref path. Per OpenAPI spec, mapping values can be either: - Full refs: "#/components/schemas/Pet" or "./other.yaml#/components/schemas/Pet" - Short names: "Pet" or "Pet.V1" (relative to #/components/schemas/) - Relative paths: "schemas/Pet" or "./other.yaml" Values containing "/" or "#" are treated as paths/refs and passed through. All other values (including those with dots like "Pet.V1") are treated as short schema names and normalized to full refs. Note: Bare file references without path separators (e.g., "other.yaml") will be treated as schema names. Use "./other.yaml" format for file references. Note: This could be a staticmethod, but @snooper_to_methods() decorator converts staticmethods to regular functions when pysnooper is installed. """ if "/" in mapping_value or "#" in mapping_value: return mapping_value return f"#/components/schemas/{mapping_value}" def _normalize_discriminator(self, discriminator: dict[str, Any]) -> dict[str, Any]: """Return a copy of the discriminator dict with normalized mapping refs.""" result = discriminator.copy() mapping = discriminator.get("mapping") if mapping: result["mapping"] = { k: self._normalize_discriminator_mapping_ref(v) for k, v in mapping.items() if isinstance(v, str) } return result def _get_discriminator_union_type(self, ref: str) -> DataType | None: """Create a union type for discriminator subtypes if available. First tries to use allOf subtypes. If none found, falls back to using the discriminator mapping to create the union type. This handles cases where schemas don't use allOf inheritance but have explicit discriminator mappings. """ subtypes = self._discriminator_subtypes.get(ref, []) if not subtypes: discriminator = self._discriminator_schemas[ref] mapping = discriminator.get("mapping", {}) if mapping: subtypes = [ self._normalize_discriminator_mapping_ref(v) for v in mapping.values() if isinstance(v, str) ] if not subtypes: return None refs = map(self.model_resolver.add_ref, subtypes) return self.data_type(data_types=[self.data_type(reference=r) for r in refs]) def get_ref_data_type(self, ref: str) -> DataType: """Get data type for a reference, handling discriminator polymorphism.""" if ref in self._discriminator_schemas and (union_type := self._get_discriminator_union_type(ref)): return union_type return super().get_ref_data_type(ref) def parse_object_fields( self, obj: JsonSchemaObject, path: list[str], module_name: Optional[str] = None, # noqa: UP045 class_name: Optional[str] = None, # noqa: UP045 ) -> list[DataModelFieldBase]: """Parse object fields, adding discriminator info for allOf polymorphism.""" fields = super().parse_object_fields(obj, path, module_name, class_name=class_name) properties = obj.properties or {} result_fields: list[DataModelFieldBase] = [] for field_obj in fields: field = properties.get(field_obj.original_name) # ty: ignore if ( isinstance(field, JsonSchemaObject) and field.ref and (discriminator := self._discriminator_schemas.get(field.ref)) ): new_field_type = self._get_discriminator_union_type(field.ref) or field_obj.data_type normalized_discriminator = self._normalize_discriminator(discriminator) field_obj = self.data_model_field_type(**{ # noqa: PLW2901 # ty: ignore **field_obj.__dict__, "data_type": new_field_type, "extras": {**field_obj.extras, "discriminator": normalized_discriminator}, }) result_fields.append(field_obj) return result_fields def resolve_object(self, obj: ReferenceObject | BaseModelT, object_type: type[BaseModelT]) -> BaseModelT: """Resolve a reference object to its actual type or return the object as-is.""" if isinstance(obj, ReferenceObject): ref_obj = self.get_ref_model(obj.ref) return object_type.model_validate(ref_obj) return obj def _parse_schema_or_ref( self, name: str, schema: JsonSchemaObject | ReferenceObject | None, path: list[str], ) -> DataType | None: """Parse a schema object or resolve a reference to get DataType.""" if schema is None: return None if isinstance(schema, JsonSchemaObject): return self.parse_schema(name, schema, path) self.resolve_ref(schema.ref) return self.get_ref_data_type(schema.ref) def _normalize_path(self, path: str) -> str: # noqa: PLR6301 """Normalize path for consistent matching. Note: This is an instance method (not static) due to the snooper_to_methods class decorator which does not preserve staticmethod descriptors. """ if not path.startswith("/"): path = f"/{path}" return path def _matches_path_pattern(self, path: str) -> bool: """Check if path matches any of the include patterns.""" if not self.openapi_include_paths: return True normalized_path = self._normalize_path(path) return any( fnmatch.fnmatch(normalized_path, self._normalize_path(pattern)) for pattern in self.openapi_include_paths ) def _process_path_items( # noqa: PLR0913 self, items: dict[str, dict[str, Any]], base_path: list[str], scope_name: str, global_parameters: list[dict[str, Any]], security: list[dict[str, list[str]]] | None, *, strip_leading_slash: bool = True, apply_path_filter: bool = True, ) -> None: """Process path or webhook items with operations.""" scope_path = [*base_path, f"#/{scope_name}"] for item_name, methods_ in items.items(): if apply_path_filter and not self._matches_path_pattern(item_name): continue item_ref = methods_.get("$ref") if item_ref: methods = self.get_ref_model(item_ref) # Extract base path from reference for external file resolution resolved_ref = self.model_resolver.resolve_ref(item_ref) ref_file = resolved_ref.split("#")[0] if "#" in resolved_ref else resolved_ref ref_base_path = Path(ref_file).parent if ref_file and not is_url(ref_file) else None else: methods = methods_ ref_base_path = None item_parameters = global_parameters.copy() if "parameters" in methods: item_parameters.extend(methods["parameters"]) relative_name = item_name[1:] if strip_leading_slash else item_name.removeprefix("/") path = [*scope_path, relative_name] if relative_name else get_special_path("root", scope_path) base_path_context = ( self.model_resolver.current_base_path_context(ref_base_path) if ref_base_path else nullcontext() ) with base_path_context: for operation_name, raw_operation in methods.items(): if operation_name not in OPERATION_NAMES: continue if item_parameters: if "parameters" in raw_operation: raw_operation["parameters"].extend(item_parameters) else: raw_operation["parameters"] = item_parameters.copy() if security is not None and "security" not in raw_operation: raw_operation["security"] = security self.parse_operation(raw_operation, [*path, operation_name]) def parse_schema( self, name: str, obj: JsonSchemaObject, path: list[str], ) -> DataType: """Parse a JSON schema object into a data type.""" if obj.is_array: data_type = self.parse_array(name, obj, [*path, name]) elif obj.allOf: # pragma: no cover data_type = self.parse_all_of(name, obj, path) elif obj.oneOf or obj.anyOf: # pragma: no cover data_type = self.parse_root_type(name, obj, path) if isinstance(data_type, EmptyDataType) and obj.properties: self.parse_object(name, obj, path) elif obj.is_object: data_type = self.parse_object(name, obj, path) elif obj.enum and not self.ignore_enum_constraints: # pragma: no cover data_type = self.parse_enum(name, obj, path) elif obj.ref: # pragma: no cover data_type = self.get_ref_data_type(obj.ref) else: data_type = self.get_data_type(obj) self.parse_ref(obj, path) return data_type def parse_request_body( self, name: str, request_body: RequestBodyObject, path: list[str], ) -> dict[str, DataType]: """Parse request body content into data types by media type.""" data_types: dict[str, DataType] = {} for media_type, media_obj in request_body.content.items(): data_type = self._parse_schema_or_ref(name, media_obj.schema_, [*path, media_type]) if data_type: data_types[media_type] = data_type return data_types def parse_responses( self, name: str, responses: dict[str | int, ReferenceObject | ResponseObject], path: list[str], ) -> dict[str | int, dict[str, DataType]]: """Parse response objects into data types by status code and content type.""" data_types: defaultdict[str | int, dict[str, DataType]] = defaultdict(dict) for status_code, detail in responses.items(): response_name = f"{name}{str(status_code).capitalize()}" if self.use_status_code_in_response_name else name if isinstance(detail, ReferenceObject): if not detail.ref: # pragma: no cover continue ref_model = self.get_ref_model(detail.ref) content = {k: MediaObject.model_validate(v) for k, v in ref_model.get("content", {}).items()} else: content = detail.content if self.allow_responses_without_content and not content: data_types[status_code]["application/json"] = DataType(type="None") for content_type, obj in content.items(): response_path: list[str] = [*path, str(status_code), str(content_type)] data_type = self._parse_schema_or_ref(response_name, obj.schema_, response_path) if data_type: data_types[status_code][content_type] = data_type # ty: ignore return data_types @classmethod def parse_tags( cls, name: str, # noqa: ARG003 tags: list[str], path: list[str], # noqa: ARG003 ) -> list[str]: """Parse operation tags.""" return tags _field_name_resolver: FieldNameResolver = FieldNameResolver() @classmethod def _get_model_name(cls, path_name: str, method: str, suffix: str) -> str: normalized = cls._field_name_resolver.get_valid_name(path_name, ignore_snake_case_field=True) camel_path_name = snake_to_upper_camel(normalized) return f"{camel_path_name}{method.capitalize()}{suffix}" def parse_all_parameters( # noqa: PLR0912, PLR0914, PLR0915 self, name: str, parameters: list[ReferenceObject | ParameterObject], path: list[str], ) -> DataType | None: """Parse all operation parameters into a data model.""" fields: list[DataModelFieldBase] = [] exclude_field_names: set[str] = set() seen_parameter_names: set[str] = set() reference = self.model_resolver.add(path, name, class_name=True, unique=True) for parameter_ in parameters: parameter = self.resolve_object(parameter_, ParameterObject) parameter_name = parameter.name if ( not parameter_name or parameter.in_ not in {ParameterLocation.query, ParameterLocation.path} or (parameter.in_ == ParameterLocation.path and not self.include_path_parameters) ): continue if parameter_name in seen_parameter_names: msg = f"Parameter name '{parameter_name}' is used more than once." raise Exception(msg) # noqa: TRY002 seen_parameter_names.add(parameter_name) field_name, alias = self.model_resolver.get_valid_field_name_and_alias( field_name=parameter_name, excludes=exclude_field_names, model_type=self.field_name_model_type, class_name=name, ) if parameter.schema_: param_schema = parameter.schema_ if param_schema.has_ref_with_schema_keywords and not param_schema.is_ref_with_nullable_only: param_schema = self._merge_ref_with_schema(param_schema) effective_default, effective_has_default = self.model_resolver.resolve_default_value( parameter_name, param_schema.default, param_schema.has_default, class_name=reference.name, ) effective_required = parameter.required if self.apply_default_values_for_required_fields and effective_has_default: effective_required = False fields.append( self.get_object_field( field_name=field_name, field=param_schema, field_type=self.parse_item(field_name, param_schema, [*path, name, parameter_name]), original_field_name=parameter_name, required=effective_required, alias=alias, effective_default=effective_default, effective_has_default=effective_has_default, ) ) else: data_types: list[DataType] = [] object_schema: JsonSchemaObject | None = None for ( media_type, media_obj, ) in parameter.content.items(): if not media_obj.schema_: continue object_schema = self.resolve_object(media_obj.schema_, JsonSchemaObject) data_types.append( self.parse_item( field_name, object_schema, [*path, name, parameter_name, media_type], ) ) if not data_types: continue if len(data_types) == 1: data_type = data_types[0] else: data_type = self.data_type(data_types=data_types) # multiple data_type parse as non-constraints field object_schema = None original_default = object_schema.default if object_schema else None original_has_default = object_schema.has_default if object_schema else False effective_default, effective_has_default = self.model_resolver.resolve_default_value( parameter_name, original_default, original_has_default, class_name=reference.name, ) effective_required = parameter.required if self.apply_default_values_for_required_fields and effective_has_default: effective_required = False # Handle multiple aliases (Pydantic v2 AliasChoices) single_alias: str | None = None validation_aliases: list[str] | None = None if isinstance(alias, list): validation_aliases = alias else: single_alias = alias fields.append( self.data_model_field_type( name=field_name, default=effective_default, data_type=data_type, required=effective_required, alias=single_alias, validation_aliases=validation_aliases, constraints=object_schema.model_dump(exclude_none=True) if object_schema and self.is_constraints_field(object_schema) else None, nullable=object_schema.nullable if object_schema and self.strict_nullable and object_schema.nullable is not None else ( False if object_schema and self.strict_nullable and (effective_has_default or effective_required) else None ), strip_default_none=self.strip_default_none, extras=self.get_field_extras(object_schema) if object_schema else {}, use_annotated=self.use_annotated, use_serialize_as_any=self.use_serialize_as_any, use_field_description=self.use_field_description, use_field_description_example=self.use_field_description_example, use_inline_field_description=self.use_inline_field_description, use_default_kwarg=self.use_default_kwarg, original_name=parameter_name, has_default=effective_has_default, type_has_null=object_schema.type_has_null if object_schema else None, use_serialization_alias=self.use_serialization_alias, ) ) if OpenAPIScope.Parameters in self.open_api_scopes and fields: # Using _create_data_model from parent class JsonSchemaParser # This method automatically adds frozen=True for DataClass types self.results.append( self._create_data_model( fields=fields, reference=reference, custom_base_class=self._resolve_base_class(name), custom_template_dir=self.custom_template_dir, extra_template_data=self.extra_template_data, keyword_only=self.keyword_only, treat_dot_as_module=self.treat_dot_as_module, dataclass_arguments=self.dataclass_arguments, ) ) return self.data_type(reference=reference) return None def parse_operation( self, raw_operation: dict[str, Any], path: list[str], ) -> None: """Parse an OpenAPI operation including parameters, request body, and responses.""" operation = Operation.model_validate(raw_operation) path_name, method = path[-2:] if self.use_operation_id_as_name: if not operation.operationId: msg = ( f"All operations must have an operationId when --use_operation_id_as_name is set." f"The following path was missing an operationId: {path_name}" ) raise Error(msg) path_name = operation.operationId method = "" self.parse_all_parameters( self._get_model_name( path_name, method, suffix="Parameters" if self.include_path_parameters else "ParametersQuery" ), operation.parameters, [*path, "parameters"], ) if operation.requestBody: if isinstance(operation.requestBody, ReferenceObject): ref_model = self.get_ref_model(operation.requestBody.ref) request_body = RequestBodyObject.model_validate(ref_model) else: request_body = operation.requestBody self.parse_request_body( name=self._get_model_name(path_name, method, suffix="Request"), request_body=request_body, path=[*path, "requestBody"], ) self.parse_responses( name=self._get_model_name(path_name, method, suffix="Response"), responses=operation.responses, path=[*path, "responses"], ) if OpenAPIScope.Tags in self.open_api_scopes: self.parse_tags( name=self._get_model_name(path_name, method, suffix="Tags"), tags=operation.tags, path=[*path, "tags"], ) def parse_raw(self) -> None: # noqa: PLR0912 """Parse OpenAPI specification including schemas, paths, and operations.""" for source, path_parts in self._get_context_source_path_parts(): if self.validation: warn( "Deprecated: `--validation` option is deprecated. the option will be removed in a future " "release. please use another tool to validate OpenAPI.\n", stacklevel=2, ) if source.raw_data is not None: warn( "Warning: Validation was skipped for dict input. " "The --validation option only works with file or text input.\n", stacklevel=2, ) else: try: from prance import BaseParser # noqa: PLC0415 BaseParser( spec_string=source.text, backend="openapi-spec-validator", encoding=self.encoding, ) except ImportError: # pragma: no cover warn( "Warning: Validation was skipped for OpenAPI. " "`prance` or `openapi-spec-validator` are not installed.\n" "To use --validation option after datamodel-code-generator 0.24.0, " "Please run `$pip install 'datamodel-code-generator[validation]'`.\n", stacklevel=2, ) specification: dict[str, Any] = ( dict(source.raw_data) if source.raw_data is not None else load_data(source.text) ) self.raw_obj = specification self._collect_discriminator_schemas() schemas: dict[str, Any] = specification.get("components", {}).get("schemas", {}) paths: dict[str, Any] = specification.get("paths", {}) security: list[dict[str, list[str]]] | None = specification.get("security") # Warn if schemas is empty but paths exist and only Schemas scope is used if not schemas and self.open_api_scopes == [OpenAPIScope.Schemas] and paths: warn( "No schemas found in components/schemas. If your schemas are defined in " "external files referenced from paths, consider using --openapi-scopes paths", stacklevel=2, ) if OpenAPIScope.Schemas in self.open_api_scopes: for obj_name, raw_obj in schemas.items(): self.parse_raw_obj( obj_name, raw_obj, [*path_parts, "#/components", "schemas", obj_name], ) if OpenAPIScope.Paths in self.open_api_scopes: # Resolve $ref in global parameter list global_parameters = [ self._get_ref_body(p["$ref"]) if isinstance(p, dict) and "$ref" in p else p for p in paths.get("parameters", []) if isinstance(p, dict) ] self._process_path_items(paths, path_parts, "paths", global_parameters, security) if OpenAPIScope.Webhooks in self.open_api_scopes: webhooks: dict[str, dict[str, Any]] = specification.get("webhooks", {}) self._process_path_items( webhooks, path_parts, "webhooks", [], security, strip_leading_slash=False, apply_path_filter=False ) if OpenAPIScope.RequestBodies in self.open_api_scopes: request_bodies: dict[str, Any] = specification.get("components", {}).get("requestBodies", {}) for body_name, raw_body in request_bodies.items(): resolved_body = self.get_ref_model(raw_body["$ref"]) if "$ref" in raw_body else raw_body content = resolved_body.get("content", {}) for media_type, media_obj in content.items(): schema = media_obj.get("schema") if not schema: continue self.parse_raw_obj( body_name, schema, [ *path_parts, "#/components", "requestBodies", body_name, "content", media_type, "schema", ], ) self._resolve_unparsed_json_pointer() self._generate_forced_base_models() def _collect_discriminator_schemas(self) -> None: """Collect schemas with discriminators but no oneOf/anyOf, and find their subtypes.""" schemas: dict[str, Any] = self.raw_obj.get("components", {}).get("schemas", {}) potential_subtypes: dict[str, list[str]] = {} for schema_name, schema in schemas.items(): discriminator = schema.get("discriminator") if discriminator and not schema.get("oneOf") and not schema.get("anyOf"): ref = f"#/components/schemas/{schema_name}" self._discriminator_schemas[ref] = discriminator all_of = schema.get("allOf") if all_of: refs = [item.get("$ref") for item in all_of if item.get("$ref")] if refs: potential_subtypes[schema_name] = refs for schema_name, refs in potential_subtypes.items(): for ref_in_allof in refs: if ref_in_allof in self._discriminator_schemas: subtype_ref = f"#/components/schemas/{schema_name}" self._discriminator_subtypes[ref_in_allof].append(subtype_ref)