from __future__ import annotations import re from collections import defaultdict from enum import Enum from pathlib import Path from typing import ( Any, Callable, ClassVar, DefaultDict, Dict, Iterable, List, Mapping, Optional, Pattern, Sequence, Set, Tuple, Type, TypeVar, Union, ) from urllib.parse import ParseResult from warnings import warn from pydantic import Field from datamodel_code_generator import ( DefaultPutDict, Error, LiteralType, OpenAPIScope, PythonVersion, load_yaml, snooper_to_methods, ) from datamodel_code_generator.format import DatetimeClassType from datamodel_code_generator.model import DataModel, DataModelFieldBase from datamodel_code_generator.model import pydantic as pydantic_model 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 snake_to_upper_camel from datamodel_code_generator.types import ( DataType, DataTypeManager, EmptyDataType, StrictTypes, ) from datamodel_code_generator.util import BaseModel 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): query = 'query' header = 'header' path = 'path' cookie = 'cookie' BaseModelT = TypeVar('BaseModelT', bound=BaseModel) class ReferenceObject(BaseModel): ref: str = Field(..., alias='$ref') class ExampleObject(BaseModel): summary: Optional[str] = None description: Optional[str] = None value: Any = None externalValue: Optional[str] = None class MediaObject(BaseModel): schema_: Union[ReferenceObject, JsonSchemaObject, None] = Field( None, alias='schema' ) example: Any = None examples: Union[str, ReferenceObject, ExampleObject, None] = None class ParameterObject(BaseModel): name: Optional[str] = None in_: Optional[ParameterLocation] = Field(None, alias='in') description: Optional[str] = None required: bool = False deprecated: bool = False schema_: Optional[JsonSchemaObject] = Field(None, alias='schema') example: Any = None examples: Union[str, ReferenceObject, ExampleObject, None] = None content: Dict[str, MediaObject] = {} class HeaderObject(BaseModel): description: Optional[str] = None required: bool = False deprecated: bool = False schema_: Optional[JsonSchemaObject] = Field(None, alias='schema') example: Any = None examples: Union[str, ReferenceObject, ExampleObject, None] = None content: Dict[str, MediaObject] = {} class RequestBodyObject(BaseModel): description: Optional[str] = None content: Dict[str, MediaObject] = {} required: bool = False class ResponseObject(BaseModel): description: Optional[str] = None headers: Dict[str, ParameterObject] = {} content: Dict[Union[str, int], MediaObject] = {} class Operation(BaseModel): tags: List[str] = [] summary: Optional[str] = None description: Optional[str] = None operationId: Optional[str] = None parameters: List[Union[ReferenceObject, ParameterObject]] = [] requestBody: Union[ReferenceObject, RequestBodyObject, None] = None responses: Dict[Union[str, int], Union[ReferenceObject, ResponseObject]] = {} deprecated: bool = False class ComponentsObject(BaseModel): schemas: Dict[str, Union[ReferenceObject, JsonSchemaObject]] = {} responses: Dict[str, Union[ReferenceObject, ResponseObject]] = {} examples: Dict[str, Union[ReferenceObject, ExampleObject]] = {} requestBodies: Dict[str, Union[ReferenceObject, RequestBodyObject]] = {} headers: Dict[str, Union[ReferenceObject, HeaderObject]] = {} @snooper_to_methods(max_variable_length=None) class OpenAPIParser(JsonSchemaParser): SCHEMA_PATHS: ClassVar[List[str]] = ['#/components/schemas'] def __init__( self, source: Union[str, Path, List[Path], ParseResult], *, data_model_type: Type[DataModel] = pydantic_model.BaseModel, data_model_root_type: Type[DataModel] = pydantic_model.CustomRootType, data_type_manager_type: Type[DataTypeManager] = pydantic_model.DataTypeManager, data_model_field_type: Type[DataModelFieldBase] = pydantic_model.DataModelField, base_class: Optional[str] = None, additional_imports: Optional[List[str]] = None, custom_template_dir: Optional[Path] = None, extra_template_data: Optional[DefaultDict[str, Dict[str, Any]]] = None, target_python_version: PythonVersion = PythonVersion.PY_38, dump_resolve_reference_action: Optional[Callable[[Iterable[str]], str]] = None, validation: bool = False, field_constraints: bool = False, snake_case_field: bool = False, strip_default_none: bool = False, aliases: Optional[Mapping[str, str]] = None, allow_population_by_field_name: bool = False, allow_extra_fields: bool = False, apply_default_values_for_required_fields: bool = False, force_optional_for_required_fields: bool = False, class_name: Optional[str] = None, use_standard_collections: bool = False, base_path: Optional[Path] = None, use_schema_description: bool = False, use_field_description: bool = False, use_default_kwarg: bool = False, reuse_model: bool = False, encoding: str = 'utf-8', enum_field_as_literal: Optional[LiteralType] = None, use_one_literal_as_default: bool = False, set_default_enum_member: bool = False, use_subclass_enum: bool = False, strict_nullable: bool = False, use_generic_container_types: bool = False, enable_faux_immutability: bool = False, remote_text_cache: Optional[DefaultPutDict[str, str]] = None, disable_appending_item_suffix: bool = False, strict_types: Optional[Sequence[StrictTypes]] = None, empty_enum_field_name: Optional[str] = None, custom_class_name_generator: Optional[Callable[[str], str]] = None, field_extra_keys: Optional[Set[str]] = None, field_include_all_keys: bool = False, field_extra_keys_without_x_prefix: Optional[Set[str]] = None, openapi_scopes: Optional[List[OpenAPIScope]] = None, wrap_string_literal: Optional[bool] = False, use_title_as_name: bool = False, use_operation_id_as_name: bool = False, use_unique_items_as_set: bool = False, http_headers: Optional[Sequence[Tuple[str, str]]] = None, http_ignore_tls: bool = False, use_annotated: bool = False, use_non_positive_negative_number_constrained_types: bool = False, original_field_name_delimiter: Optional[str] = None, use_double_quotes: bool = False, use_union_operator: bool = False, allow_responses_without_content: bool = False, collapse_root_models: bool = False, special_field_name_prefix: Optional[str] = None, remove_special_field_name_prefix: bool = False, capitalise_enum_members: bool = False, keep_model_order: bool = False, known_third_party: Optional[List[str]] = None, custom_formatters: Optional[List[str]] = None, custom_formatters_kwargs: Optional[Dict[str, Any]] = None, use_pendulum: bool = False, http_query_parameters: Optional[Sequence[Tuple[str, str]]] = None, treat_dots_as_module: bool = False, use_exact_imports: bool = False, default_field_extras: Optional[Dict[str, Any]] = None, target_datetime_class: DatetimeClassType = DatetimeClassType.Datetime, keyword_only: bool = False, no_alias: bool = False, ): super().__init__( source=source, data_model_type=data_model_type, data_model_root_type=data_model_root_type, data_type_manager_type=data_type_manager_type, data_model_field_type=data_model_field_type, base_class=base_class, additional_imports=additional_imports, custom_template_dir=custom_template_dir, extra_template_data=extra_template_data, target_python_version=target_python_version, dump_resolve_reference_action=dump_resolve_reference_action, validation=validation, field_constraints=field_constraints, snake_case_field=snake_case_field, strip_default_none=strip_default_none, aliases=aliases, allow_population_by_field_name=allow_population_by_field_name, allow_extra_fields=allow_extra_fields, apply_default_values_for_required_fields=apply_default_values_for_required_fields, force_optional_for_required_fields=force_optional_for_required_fields, class_name=class_name, use_standard_collections=use_standard_collections, base_path=base_path, use_schema_description=use_schema_description, use_field_description=use_field_description, use_default_kwarg=use_default_kwarg, reuse_model=reuse_model, encoding=encoding, enum_field_as_literal=enum_field_as_literal, use_one_literal_as_default=use_one_literal_as_default, set_default_enum_member=set_default_enum_member, use_subclass_enum=use_subclass_enum, strict_nullable=strict_nullable, use_generic_container_types=use_generic_container_types, enable_faux_immutability=enable_faux_immutability, remote_text_cache=remote_text_cache, disable_appending_item_suffix=disable_appending_item_suffix, strict_types=strict_types, empty_enum_field_name=empty_enum_field_name, custom_class_name_generator=custom_class_name_generator, field_extra_keys=field_extra_keys, field_include_all_keys=field_include_all_keys, field_extra_keys_without_x_prefix=field_extra_keys_without_x_prefix, wrap_string_literal=wrap_string_literal, use_title_as_name=use_title_as_name, use_operation_id_as_name=use_operation_id_as_name, use_unique_items_as_set=use_unique_items_as_set, http_headers=http_headers, http_ignore_tls=http_ignore_tls, use_annotated=use_annotated, use_non_positive_negative_number_constrained_types=use_non_positive_negative_number_constrained_types, original_field_name_delimiter=original_field_name_delimiter, use_double_quotes=use_double_quotes, use_union_operator=use_union_operator, allow_responses_without_content=allow_responses_without_content, collapse_root_models=collapse_root_models, special_field_name_prefix=special_field_name_prefix, remove_special_field_name_prefix=remove_special_field_name_prefix, capitalise_enum_members=capitalise_enum_members, keep_model_order=keep_model_order, known_third_party=known_third_party, custom_formatters=custom_formatters, custom_formatters_kwargs=custom_formatters_kwargs, use_pendulum=use_pendulum, http_query_parameters=http_query_parameters, treat_dots_as_module=treat_dots_as_module, use_exact_imports=use_exact_imports, default_field_extras=default_field_extras, target_datetime_class=target_datetime_class, keyword_only=keyword_only, no_alias=no_alias, ) self.open_api_scopes: List[OpenAPIScope] = openapi_scopes or [ OpenAPIScope.Schemas ] def get_ref_model(self, ref: str) -> Dict[str, Any]: ref_file, ref_path = self.model_resolver.resolve_ref(ref).split('#', 1) if ref_file: ref_body = self._get_ref_body(ref_file) else: # pragma: no cover ref_body = self.raw_obj return get_model_by_path(ref_body, ref_path.split('/')[1:]) def get_data_type(self, obj: JsonSchemaObject) -> DataType: # OpenAPI 3.0 doesn't allow `null` in the `type` field and list of types # https://swagger.io/docs/specification/data-models/data-types/#null # OpenAPI 3.1 does allow `null` in the `type` field and is equivalent to # a `nullable` flag on the property itself if obj.nullable and self.strict_nullable and isinstance(obj.type, str): obj.type = [obj.type, 'null'] return super().get_data_type(obj) def resolve_object( self, obj: Union[ReferenceObject, BaseModelT], object_type: Type[BaseModelT] ) -> BaseModelT: if isinstance(obj, ReferenceObject): ref_obj = self.get_ref_model(obj.ref) return object_type.parse_obj(ref_obj) return obj def parse_schema( self, name: str, obj: JsonSchemaObject, path: List[str], ) -> DataType: 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: # 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], ) -> None: for ( media_type, media_obj, ) in request_body.content.items(): # type: str, MediaObject if isinstance(media_obj.schema_, JsonSchemaObject): self.parse_schema(name, media_obj.schema_, [*path, media_type]) def parse_responses( self, name: str, responses: Dict[Union[str, int], Union[ReferenceObject, ResponseObject]], path: List[str], ) -> Dict[Union[str, int], Dict[str, DataType]]: data_types: DefaultDict[Union[str, int], Dict[str, DataType]] = defaultdict( dict ) for status_code, detail in responses.items(): if isinstance(detail, ReferenceObject): if not detail.ref: # pragma: no cover continue ref_model = self.get_ref_model(detail.ref) content = { k: MediaObject.parse_obj(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(): object_schema = obj.schema_ if not object_schema: # pragma: no cover continue if isinstance(object_schema, JsonSchemaObject): data_types[status_code][content_type] = self.parse_schema( name, object_schema, [*path, str(status_code), content_type] ) else: data_types[status_code][content_type] = self.get_ref_data_type( object_schema.ref ) return data_types @classmethod def parse_tags( cls, name: str, tags: List[str], path: List[str], ) -> List[str]: return tags @classmethod def _get_model_name(cls, path_name: str, method: str, suffix: str) -> str: camel_path_name = snake_to_upper_camel(path_name.replace('/', '_')) return f'{camel_path_name}{method.capitalize()}{suffix}' def parse_all_parameters( self, name: str, parameters: List[Union[ReferenceObject, ParameterObject]], path: List[str], ) -> None: fields: List[DataModelFieldBase] = [] exclude_field_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_ != ParameterLocation.query: continue field_name, alias = self.model_resolver.get_valid_field_name_and_alias( field_name=parameter_name, excludes=exclude_field_names ) if parameter.schema_: fields.append( self.get_object_field( field_name=field_name, field=parameter.schema_, field_type=self.parse_item( field_name, parameter.schema_, [*path, name, parameter_name] ), original_field_name=parameter_name, required=parameter.required, alias=alias, ) ) else: data_types: List[DataType] = [] object_schema: Optional[JsonSchemaObject] = 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 fields.append( self.data_model_field_type( name=field_name, default=object_schema.default if object_schema else None, data_type=data_type, required=parameter.required, alias=alias, constraints=object_schema.dict() 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.has_default or parameter.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_field_description=self.use_field_description, use_default_kwarg=self.use_default_kwarg, original_name=parameter_name, has_default=object_schema.has_default if object_schema else False, type_has_null=object_schema.type_has_null if object_schema else None, ) ) if OpenAPIScope.Parameters in self.open_api_scopes and fields: self.results.append( self.data_model_type( fields=fields, reference=reference, custom_base_class=self.base_class, custom_template_dir=self.custom_template_dir, keyword_only=self.keyword_only, ) ) def parse_operation( self, raw_operation: Dict[str, Any], path: List[str], ) -> None: operation = Operation.parse_obj(raw_operation) path_name, method = path[-2:] if self.use_operation_id_as_name: if not operation.operationId: raise Error( 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}' ) path_name = operation.operationId method = '' self.parse_all_parameters( self._get_model_name(path_name, method, suffix='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.parse_obj(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: 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' ) try: from prance import BaseParser 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" ) specification: Dict[str, Any] = load_yaml(source.text) self.raw_obj = specification schemas: Dict[Any, Any] = specification.get('components', {}).get( 'schemas', {} ) security: Optional[List[Dict[str, List[str]]]] = specification.get( 'security' ) if OpenAPIScope.Schemas in self.open_api_scopes: for ( obj_name, raw_obj, ) in schemas.items(): # type: str, Dict[Any, Any] self.parse_raw_obj( obj_name, raw_obj, [*path_parts, '#/components', 'schemas', obj_name], ) if OpenAPIScope.Paths in self.open_api_scopes: paths: Dict[str, Dict[str, Any]] = specification.get('paths', {}) parameters: List[Dict[str, Any]] = [ self._get_ref_body(p['$ref']) if '$ref' in p else p for p in paths.get('parameters', []) if isinstance(p, dict) ] paths_path = [*path_parts, '#/paths'] for path_name, methods in paths.items(): # Resolve path items if applicable if '$ref' in methods: methods = self.get_ref_model(methods['$ref']) paths_parameters = parameters[:] if 'parameters' in methods: paths_parameters.extend(methods['parameters']) relative_path_name = path_name[1:] if relative_path_name: path = [*paths_path, relative_path_name] else: # pragma: no cover path = get_special_path('root', paths_path) for operation_name, raw_operation in methods.items(): if operation_name not in OPERATION_NAMES: continue if paths_parameters: if 'parameters' in raw_operation: # pragma: no cover raw_operation['parameters'].extend(paths_parameters) else: raw_operation['parameters'] = paths_parameters if security is not None and 'security' not in raw_operation: raw_operation['security'] = security self.parse_operation( raw_operation, [*path, operation_name], ) self._resolve_unparsed_json_pointer()