Skip to content
Merged
12 changes: 12 additions & 0 deletions src/datamodel_code_generator/model/_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,15 @@ class WrappedDefault:
def __repr__(self) -> str:
"""Return type constructor representation, e.g., 'CountType(10)'."""
return f"{self.type_name}({self.value!r})"


@dataclass(repr=False)
class ValidatedDefault:
"""Represents a structured default validated through the field type."""

value: Any
type_name: str

def __repr__(self) -> str:
"""Return TypeAdapter validation representation for structured defaults."""
return f"TypeAdapter({self.type_name}).validate_python({self.value!r})"
5 changes: 3 additions & 2 deletions src/datamodel_code_generator/model/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
IMPORT_UNION,
Import,
)
from datamodel_code_generator.model._types import WrappedDefault
from datamodel_code_generator.model._types import ValidatedDefault, WrappedDefault
from datamodel_code_generator.reference import Reference, _BaseModel
from datamodel_code_generator.types import (
ANY,
Expand All @@ -38,7 +38,7 @@
get_optional_type,
)

__all__ = ["WrappedDefault"]
__all__ = ["ValidatedDefault", "WrappedDefault"]

if TYPE_CHECKING:
from collections.abc import Iterator
Expand Down Expand Up @@ -599,6 +599,7 @@ class DataModel(TemplateBase, Nullable, ABC): # noqa: PLR0904
SUPPORTS_DISCRIMINATOR: ClassVar[bool] = False
SUPPORTS_FIELD_RENAMING: ClassVar[bool] = False
SUPPORTS_WRAPPED_DEFAULT: ClassVar[bool] = False
SUPPORTS_VALIDATED_DEFAULT: ClassVar[bool] = False
SUPPORTS_KW_ONLY: ClassVar[bool] = False
has_forward_reference: bool = False

Expand Down
13 changes: 9 additions & 4 deletions src/datamodel_code_generator/model/pydantic_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,15 @@
DataModel,
DataModelFieldBase,
)
from datamodel_code_generator.model._types import WrappedDefault
from datamodel_code_generator.model._types import ValidatedDefault, WrappedDefault
from datamodel_code_generator.model.base import UNDEFINED, repr_set_sorted
from datamodel_code_generator.types import STANDARD_LIST, UnionIntFloat, chain_as_tuple

# Defined here instead of importing from pydantic_v2.imports to avoid circular import
# (pydantic_base -> pydantic_v2.imports -> pydantic_v2/__init__ -> pydantic_v2.base_model -> pydantic_base)
IMPORT_ANYURL = Import.from_full_path("pydantic.AnyUrl")
IMPORT_FIELD = Import.from_full_path("pydantic.Field")
IMPORT_TYPE_ADAPTER = Import.from_full_path("pydantic.TypeAdapter")

if TYPE_CHECKING:
from collections import defaultdict
Expand Down Expand Up @@ -126,7 +127,7 @@ def _get_strict_field_constraint_value(self, constraint: str, value: Any) -> Any
return int(value)

def _get_default_as_pydantic_model(self) -> str | None: # noqa: PLR0911, PLR0912
if isinstance(self.default, WrappedDefault):
if isinstance(self.default, (ValidatedDefault, WrappedDefault)):
return f"lambda :{self.default!r}"
if self.data_type.is_list and len(self.data_type.data_types) == 1:
data_type_child = self.data_type.data_types[0]
Expand Down Expand Up @@ -296,8 +297,12 @@ def annotated(self) -> str | None:
@property
def imports(self) -> tuple[Import, ...]:
"""Get all required imports including Field if needed."""
if self.field:
return chain_as_tuple(super().imports, (IMPORT_FIELD,))
field = self.field
if field:
extra_imports = [IMPORT_FIELD]
if isinstance(self.default, ValidatedDefault):
extra_imports.append(IMPORT_TYPE_ADAPTER)
return chain_as_tuple(super().imports, extra_imports)
return super().imports


Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,7 @@ class BaseModel(BaseModelBase):
SUPPORTS_DISCRIMINATOR: ClassVar[bool] = True
SUPPORTS_FIELD_RENAMING: ClassVar[bool] = True
SUPPORTS_WRAPPED_DEFAULT: ClassVar[bool] = True
SUPPORTS_VALIDATED_DEFAULT: ClassVar[bool] = True
# In Pydantic 2.11+, populate_by_name is deprecated in favor of validate_by_name + validate_by_alias
# Default to V2 compatible (populate_by_name) unless target_pydantic_version is specified
_CONFIG_ATTRIBUTES_V2: ClassVar[list[ConfigAttribute]] = [
Expand Down
112 changes: 111 additions & 1 deletion src/datamodel_code_generator/parser/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@
ConstraintsBase,
DataModel,
DataModelFieldBase,
ValidatedDefault,
WrappedDefault,
)
from datamodel_code_generator.model.enum import Enum, Member
Expand All @@ -82,7 +83,7 @@
from datamodel_code_generator.parser._graph import stable_toposort
from datamodel_code_generator.parser._scc import find_circular_sccs, strongly_connected_components
from datamodel_code_generator.reference import ModelResolver, ModelType, Reference
from datamodel_code_generator.types import ANY, DataType, DataTypeManager
from datamodel_code_generator.types import ANY, NONE, DataType, DataTypeManager, _remove_none_from_union
Comment thread Dismissed
from datamodel_code_generator.util import camel_to_snake

if TYPE_CHECKING:
Expand Down Expand Up @@ -399,6 +400,92 @@
yield model, field, data_type


def _unwrap_type_alias(data_type: DataType) -> DataType:
current = data_type
seen: set[int] = set()
while current.reference and isinstance(current.reference.source, TypeAliasBase):
source = current.reference.source
if id(source) in seen or not source.fields:
break
seen.add(id(source))
current = source.fields[0].data_type
return current


def _contains_model_reference(data_type: DataType) -> bool:
stack = [data_type]
seen: set[int] = set()

while stack:
resolved = _unwrap_type_alias(stack.pop())
resolved_id = id(resolved)
if resolved_id in seen:
continue
seen.add(resolved_id)

if resolved.reference and isinstance(
resolved.reference.source,
(pydantic_model_v2.BaseModel, pydantic_model_v2.RootModel),
):
return True

if resolved.dict_key:
stack.append(resolved.dict_key)
stack.extend(resolved.data_types)

return False


def _uses_existing_model_factory_path(data_type: DataType) -> bool:
for candidate in data_type.data_types or (data_type,):
if candidate.reference and isinstance(
candidate.reference.source,
(pydantic_model_v2.BaseModel, pydantic_model_v2.RootModel),
):
return True
if (
candidate.is_list
and len(candidate.data_types) == 1
and candidate.data_types[0].reference
and isinstance(
candidate.data_types[0].reference.source,
(pydantic_model_v2.BaseModel, pydantic_model_v2.RootModel),
)
):
return True
return False
Comment thread
coderabbitai[bot] marked this conversation as resolved.


def _default_matches_plain_container_branch(default: Any, data_type: DataType) -> bool:
return (isinstance(default, dict) and data_type.is_dict and not _contains_model_reference(data_type)) or (
isinstance(default, list) and data_type.is_list and not _contains_model_reference(data_type)
Comment thread
keyz marked this conversation as resolved.
Outdated
)


def _get_validated_default_type_name(data_type: DataType) -> str:
type_name = data_type.type_hint
if data_type.is_optional:
return _remove_none_from_union(type_name, use_union_operator=data_type.use_union_operator)
return type_name


def _needs_validated_default(default: Any, data_type: DataType) -> bool:
if not isinstance(default, (dict, list)):
return False

resolved = _unwrap_type_alias(data_type)
if not _contains_model_reference(resolved):
return False

if resolved.is_union:
non_none_branches = tuple(branch for branch in data_type.data_types if branch.type_hint != NONE)
Comment thread
koxudaxi marked this conversation as resolved.
Outdated
if len(non_none_branches) == 1 and _uses_existing_model_factory_path(non_none_branches[0]):
return False
return not any(_default_matches_plain_container_branch(default, branch) for branch in resolved.data_types)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

return not _uses_existing_model_factory_path(data_type)


def _alias_base_class_imports(
model: DataModel,
aliased_imports: dict[tuple[str | None, str], Import],
Expand Down Expand Up @@ -2150,6 +2237,28 @@
type_name=type_name,
)

def __wrap_validated_default_values(
self,
models: list[DataModel],
) -> None:
"""Wrap structured defaults that must be validated through the declared field type."""
if not self.data_model_type.SUPPORTS_VALIDATED_DEFAULT:
return
for model in models:
if isinstance(model, (Enum, self.data_model_root_type)):
continue
for model_field in model.fields:
if model_field.default is None:
continue
if isinstance(model_field.default, (Member, ValidatedDefault, WrappedDefault)):
continue
if not _needs_validated_default(model_field.default, model_field.data_type):
continue
model_field.default = ValidatedDefault(
value=model_field.default,
type_name=_get_validated_default_type_name(model_field.data_type),
Comment thread
keyz marked this conversation as resolved.
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

def __override_required_field(
self,
models: list[DataModel],
Expand Down Expand Up @@ -3175,6 +3284,7 @@
self.__collapse_root_models(models, unused_models, imports, scoped_model_resolver)
self.__set_default_enum_member(models)
self.__wrap_root_model_default_values(models)
self.__wrap_validated_default_values(models)
self.__sort_models(models, imports, use_deferred_annotations=config.use_deferred_annotations)
self.__change_field_name(models)
self.__apply_discriminator_type(models, imports)
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
# generated by datamodel-codegen:
# filename: type_alias_chain_model_default_object_ref.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, Field, TypeAdapter


class A(BaseModel):
type: Literal['a'] = 'a'


type Alias1 = A


type Alias2 = Alias1


class Model(BaseModel):
x: Alias2 | None = Field(
default_factory=lambda: TypeAdapter(Alias2).validate_python({'type': 'a'})
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# generated by datamodel-codegen:
# filename: type_alias_chain_union_default_object_ref.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter


class A(BaseModel):
type: Literal['a'] = 'a'


class B(BaseModel):
type: Literal['b'] = 'b'


type Alias1 = A | B


type Alias2 = Alias1


class Model(BaseModel):
x_value: Annotated[
Alias2 | None,
Field(
default_factory=lambda: TypeAdapter(Alias2).validate_python({'type': 'b'})
),
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# generated by datamodel-codegen:
# filename: type_alias_dict_union_default_object_ref.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter


class A(BaseModel):
type: Literal['a'] = 'a'


class B(BaseModel):
type: Literal['b']
value: int | None = None


type Alias = dict[str, A | B]


class Model(BaseModel):
x_value: Annotated[
Alias | None,
Field(
default_factory=lambda: TypeAdapter(Alias).validate_python(
{'k': {'type': 'b', 'value': 1}}
)
),
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
# generated by datamodel-codegen:
# filename: type_alias_inline_dict_union_default_object.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter


class A(BaseModel):
type: Literal['a'] = 'a'


class B(BaseModel):
type: Literal['b']
value: int | None = None


class Model(BaseModel):
x_value: Annotated[
dict[str, A | B] | None,
Field(
default_factory=lambda: TypeAdapter(dict[str, A | B]).validate_python(
{'k': {'type': 'b', 'value': 1}}
)
),
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
# generated by datamodel-codegen:
# filename: type_alias_inline_list_union_default_object.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter


class A(BaseModel):
type: Literal['a'] = 'a'


class B(BaseModel):
type: Literal['b']
value: int | None = None


class Model(BaseModel):
x_value: Annotated[
list[A | B] | None,
Field(
default_factory=lambda: TypeAdapter(list[A | B]).validate_python(
[{'type': 'b', 'value': 1}]
)
),
]
Loading
Loading