Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 15 additions & 8 deletions src/datamodel_code_generator/format.py
Original file line number Diff line number Diff line change
Expand Up @@ -716,10 +716,17 @@ def _is_union(node: ast.AST) -> TypeGuard[ast.Subscript]:
return isinstance(node, ast.Subscript) and _is_name_or_attr(node.value, "Union")


_CONSTRAINED_CALL_NAMES = frozenset({"conbytes", "condecimal", "confloat", "conint", "conlist", "conset", "constr"})


def _is_constrained_string_call(node: ast.AST | None) -> TypeGuard[ast.Call]:
return _is_call(node, "constr")


def _is_constrained_call(node: ast.AST | None) -> TypeGuard[ast.Call]:
return any(_is_call(node, name) for name in _CONSTRAINED_CALL_NAMES)


def _contains_constrained_string_call(node: ast.AST) -> bool:
return any(_is_constrained_string_call(child) for child in ast.walk(node))

Expand All @@ -738,7 +745,7 @@ def _is_simple_union_annotation(node: ast.AST) -> bool:
return isinstance(node, ast.Name | ast.Attribute) or (isinstance(node, ast.Constant) and node.value is None)


def _should_format_constrained_string_union(
def _should_format_constrained_call_union(
annotation: ast.AST,
value: ast.AST | None,
annotation_prefix: str,
Expand All @@ -747,9 +754,9 @@ def _should_format_constrained_string_union(
) -> TypeGuard[ast.BinOp]:
if not isinstance(annotation, ast.BinOp) or not isinstance(annotation.op, ast.BitOr) or value is None:
return False
if not (_is_constrained_string_call(annotation.left) or _is_constrained_string_call(annotation.right)):
if not (_is_constrained_call(annotation.left) or _is_constrained_call(annotation.right)):
return False
constrained_call = annotation.left if _is_constrained_string_call(annotation.left) else annotation.right
constrained_call = annotation.left if _is_constrained_call(annotation.left) else annotation.right
return (
not _is_call(value, "Field")
or len(annotation_prefix) > line_length
Expand Down Expand Up @@ -1010,7 +1017,7 @@ def _format_generated_annotation_assignment( # noqa: PLR0911, PLR0912
value_prefix = f"{annotation_prefix} = "
if statement.value is None and _is_list_of_annotated(statement.annotation):
return f"{target_prefix}{_format_list_of_annotated(statement.annotation, indent, line_length, source)}"
if _should_format_constrained_string_union(
if _should_format_constrained_call_union(
statement.annotation,
statement.value,
annotation_prefix,
Expand All @@ -1020,7 +1027,7 @@ def _format_generated_annotation_assignment( # noqa: PLR0911, PLR0912
assert statement.value is not None
value = _source_segment(source, statement.value)
return (
f"{target_prefix}{_format_constrained_string_union(statement.annotation, indent, line_length, source)}"
f"{target_prefix}{_format_constrained_call_union(statement.annotation, indent, line_length, source)}"
f" = {value}"
)
if _should_format_field_bit_or_annotation_assignment(
Expand Down Expand Up @@ -1224,21 +1231,21 @@ def _format_generated_class_statement(
return None


def _format_constrained_string_union(
def _format_constrained_call_union(
annotation: ast.BinOp,
indent: str,
line_length: int,
source: str,
) -> str:
left = annotation.left
right = annotation.right
if _is_constrained_string_call(left):
if _is_constrained_call(left):
formatted_left = _format_constrained_call(left, f"{indent} ", line_length, source)
inline_union = f"{formatted_left} | {_source_segment(source, right)}"
if "\n" not in formatted_left and len(f"{indent} {inline_union}") <= line_length:
return f"(\n{indent} {inline_union}\n{indent})"
return f"(\n{indent} {formatted_left}\n{indent} | {_source_segment(source, right)}\n{indent})"
if _is_constrained_string_call(right):
if _is_constrained_call(right):
return f"{_source_segment(source, left)} | {_format_constrained_call(right, indent, line_length, source)}"
return _source_segment(source, annotation) # pragma: no cover

Expand Down
1 change: 1 addition & 0 deletions src/datamodel_code_generator/model/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

from datamodel_code_generator import PythonVersion

from .base import UNDEFINED as UNDEFINED
from .base import ConstraintsBase, DataModel, DataModelFieldBase, _rebuild_model_with_datamodel_namespace

if TYPE_CHECKING:
Expand Down
5 changes: 2 additions & 3 deletions src/datamodel_code_generator/model/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
IMPORT_UNION,
Import,
)
from datamodel_code_generator.python_literal import represent_python_value
from datamodel_code_generator.reference import Reference, _BaseModel
from datamodel_code_generator.types import (
ANY,
Expand Down Expand Up @@ -468,9 +469,7 @@ def method(self) -> str | None:
@property
def represented_default(self) -> str:
"""Get the repr() string of the default value."""
if isinstance(self.default, set):
return repr_set_sorted(self.default)
return repr(self.default)
return represent_python_value(self.default)

@property
def annotated(self) -> str | None:
Expand Down
71 changes: 45 additions & 26 deletions src/datamodel_code_generator/model/pydantic_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,19 @@
from datamodel_code_generator import cached_path_exists
from datamodel_code_generator.imports import Import
from datamodel_code_generator.model import (
UNDEFINED,
ConstraintsBase,
DataModel,
DataModelFieldBase,
_rebuild_model_with_datamodel_namespace,
)
from datamodel_code_generator.model.base import UNDEFINED, repr_set_sorted
from datamodel_code_generator.types import UnionIntFloat, chain_as_tuple
from datamodel_code_generator.python_literal import represent_python_value
from datamodel_code_generator.types import (
UnionIntFloat,
chain_as_tuple,
merge_normalized_constraint,
normalize_integer_constraint,
)

Check notice

Code scanning / CodeQL

Cyclic import Note

Import of module
datamodel_code_generator.types
begins an import cycle.
Comment thread
koxudaxi marked this conversation as resolved.

# 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)
Expand Down Expand Up @@ -68,6 +74,7 @@
"regex",
}
_COMPARE_EXPRESSIONS: ClassVar[set[str]] = {"gt", "ge", "lt", "le"}
_INTEGER_CONSTRAINTS: ClassVar[set[str]] = _COMPARE_EXPRESSIONS | {"multiple_of"}
constraints: Optional[Constraints] = None # noqa: UP045
_PARSE_METHOD: ClassVar[str] = "model_validate"

Expand Down Expand Up @@ -138,26 +145,43 @@
return True
return bool(self.nullable and self.required and not self.use_default_with_required)

def _has_float_data_type(self) -> bool:
"""Check whether any nested data type is float-like."""
def _has_numeric_data_type(self, type_name: str, strict_import_part: str) -> bool:
"""Return whether any field data type is the given builtin or strict numeric type."""
return any(
data_type.type == "float"
or (data_type.strict and data_type.import_ and "Float" in data_type.import_.import_)
data_type.type == type_name
or (data_type.strict and data_type.import_ and strict_import_part in data_type.import_.import_)
for data_type in self.data_type.all_data_types
)

def _get_strict_field_constraint_value(self, constraint: str, value: Any, *, is_float_type: bool) -> Any:
if value is None or constraint not in self._COMPARE_EXPRESSIONS:
return value
def _get_strict_field_constraint(
self, constraint: str, value: Any, *, is_float_type: bool, is_int_type: bool
) -> tuple[str, Any] | None:
if value is None or constraint not in self._INTEGER_CONSTRAINTS:
return constraint, value
if is_float_type:
return float(value)
str_value = str(value)
if "e" in str_value.lower(): # pragma: no cover
# Scientific notation like 1e-08 - keep as float
return float(value)
if isinstance(value, int) and not isinstance(value, bool):
return value
return int(value)
return constraint, float(value)
if is_int_type:
return normalize_integer_constraint(constraint, value)
if isinstance(value, float) and value.is_integer():
return constraint, int(value)
return constraint, value

def _get_normalized_constraint_data(self) -> dict[str, Any]:
"""Build constraint data with integer-safe values, merging colliding bounds."""
assert self.constraints is not None
dumped = self.constraints._exclude_unset_dump # noqa: SLF001
has_integer_constraints = bool(self._INTEGER_CONSTRAINTS & dumped.keys())
is_float_type = has_integer_constraints and self._has_numeric_data_type("float", "Float")
is_int_type = has_integer_constraints and not is_float_type and self._has_numeric_data_type("int", "Int")
constraint_data: dict[str, Any] = {}
for k, v in dumped.items():
if (
normalized := self._get_strict_field_constraint(
k, v, is_float_type=is_float_type, is_int_type=is_int_type
)
) is not None:
merge_normalized_constraint(constraint_data, normalized[0], normalized[1])
return constraint_data

def _get_default_factory_for_optional_nested_model(self) -> str | None:
"""Get default_factory for optional nested Pydantic model fields.
Expand Down Expand Up @@ -196,12 +220,7 @@
if any(d.import_ == IMPORT_ANYURL for d in self.data_type.all_data_types):
constraint_data: dict[str, Any] = {}
else:
dumped = self.constraints._exclude_unset_dump # noqa: SLF001
is_float_type = bool(self._COMPARE_EXPRESSIONS & dumped.keys()) and self._has_float_data_type()
constraint_data = {
k: self._get_strict_field_constraint_value(k, v, is_float_type=is_float_type)
for k, v in dumped.items()
}
constraint_data = self._get_normalized_constraint_data()
data = {**data, **constraint_data}

if self.use_field_description:
Expand Down Expand Up @@ -239,7 +258,7 @@
"""Return Field() call with all constraints and metadata."""
data, default_factory = self._get_field_data_and_default_factory()

field_arguments = sorted(f"{k}={v!r}" for k, v in data.items() if v is not None)
field_arguments = sorted(f"{k}={represent_python_value(v)}" for k, v in data.items() if v is not None)

if not field_arguments and not default_factory:
if self.nullable and self.required and not self.use_default_with_required:
Expand All @@ -259,13 +278,13 @@
):
field_arguments = ["...", *field_arguments]
elif not default_factory:
default_repr = repr_set_sorted(self.default) if isinstance(self.default, set) else repr(self.default)
default_repr = represent_python_value(self.default)
field_arguments = [default_repr, *field_arguments]

if self.is_class_var:
if self.default is UNDEFINED: # pragma: no cover
return ""
return repr_set_sorted(self.default) if isinstance(self.default, set) else repr(self.default)
return represent_python_value(self.default)

return f"Field({', '.join(field_arguments)})"

Expand Down
30 changes: 22 additions & 8 deletions src/datamodel_code_generator/model/pydantic_v2/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from __future__ import annotations

import ast
from decimal import Decimal
from typing import TYPE_CHECKING, Any, ClassVar

Expand Down Expand Up @@ -65,11 +66,13 @@
IMPORT_UUID4,
IMPORT_UUID5,
)
from datamodel_code_generator.python_literal import PythonCode
from datamodel_code_generator.types import (
DataType,
StrictTypes,
Types,
UnionIntFloat,
normalize_integer_constraints,
)
from datamodel_code_generator.types import DataTypeManager as _DataTypeManagerBase

Expand Down Expand Up @@ -126,8 +129,7 @@ def type_map_factory(
IMPORT_CONSTR,
strict=StrictTypes.str in strict_types,
kwargs={
pattern_key: r"r'^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.)*"
r"([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]{0,61}[A-Za-z0-9])\Z'",
pattern_key: _get_regex_literal(HOSTNAME_REGEX),
**({"strict": True} if StrictTypes.str in strict_types else {}),
},
),
Expand Down Expand Up @@ -188,6 +190,16 @@ def strict_type_map_factory(data_type: type[DataType]) -> dict[StrictTypes, Data
)


def _get_regex_literal(pattern: str) -> str:
escaped_regex = pattern.translate(escape_characters)
raw_literal = f"r'{escaped_regex}'"
try:
ast.literal_eval(raw_literal)
except SyntaxError:
return pattern
return PythonCode(raw_literal, pattern)


class _PydanticDataTypeManager(_DataTypeManagerBase):
"""Base data type manager for Pydantic models with constrained types."""

Expand Down Expand Up @@ -299,7 +311,12 @@ def get_data_int_type( # noqa: PLR0911
return self.data_type.from_import(IMPORT_NON_NEGATIVE_INT)
if data_type_kwargs == {"le": 0} and self.use_non_positive_negative_number_constrained_types:
return self.data_type.from_import(IMPORT_NON_POSITIVE_INT)
kwargs = {k: int(v) for k, v in data_type_kwargs.items()}
if not (kwargs := normalize_integer_constraints(data_type_kwargs)):
return (
self.copy_data_type(self.strict_type_map[StrictTypes.int])
if strict
else self.copy_data_type(self.type_map[types])
)
if strict:
kwargs["strict"] = True
return self.data_type.from_import(IMPORT_CONINT, kwargs=kwargs)
Expand Down Expand Up @@ -362,9 +379,7 @@ def get_data_str_type(self, types: Types, **kwargs: Any) -> DataType:
if strict:
data_type_kwargs["strict"] = True # ty: ignore
if self.PATTERN_KEY in data_type_kwargs:
escaped_regex = data_type_kwargs[self.PATTERN_KEY].translate(escape_characters)
# TODO: remove unneeded escaped characters
data_type_kwargs[self.PATTERN_KEY] = f"r'{escaped_regex}'"
data_type_kwargs[self.PATTERN_KEY] = _get_regex_literal(data_type_kwargs[self.PATTERN_KEY])
return self.data_type.from_import(IMPORT_CONSTR, kwargs=data_type_kwargs)
if strict:
return self.copy_data_type(self.strict_type_map[StrictTypes.str])
Expand Down Expand Up @@ -523,8 +538,7 @@ def type_map_factory( # noqa: PLR0913, PLR0917
IMPORT_CONSTR,
strict=StrictTypes.str in strict_types,
kwargs={
pattern_key: r"r'^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])\.)*"
r"([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]{0,61}[A-Za-z0-9])$'",
pattern_key: _get_regex_literal(HOSTNAME_REGEX),
**({"strict": True} if StrictTypes.str in strict_types else {}),
},
),
Expand Down
53 changes: 53 additions & 0 deletions src/datamodel_code_generator/python_literal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
"""Helpers for rendering Python source literals."""

from __future__ import annotations

from math import isfinite, isnan
from typing import Any

from typing_extensions import Self


class PythonCode(str): # noqa: FURB189 - must behave as str for regex consumers.
"""Python expression rendered without extra quoting."""

code: str
__slots__ = ("code",)

def __new__(cls, code: str, value: str | None = None) -> Self:
"""Initialize with a raw Python expression and optional string value."""
obj = super().__new__(cls, code if value is None else value)
obj.code = code
return obj

def __repr__(self) -> str:
"""Render the wrapped expression."""
return self.code


def represent_python_value(value: Any) -> str: # noqa: PLR0911
Comment thread
github-advanced-security[bot] marked this conversation as resolved.
Fixed
"""Render a value as a Python expression safe for generated source."""
match value:
case PythonCode():
return value.code
case float() if isnan(value):
return "float('nan')"
case float() if not isfinite(value):
return "float('inf')" if value > 0 else "float('-inf')"
case dict():
rendered_items = ", ".join(
f"{represent_python_value(key)}: {represent_python_value(item)}" for key, item in value.items()
)
return f"{{{rendered_items}}}"
case list():
return "[" + ", ".join(represent_python_value(item) for item in value) + "]"
case tuple():
rendered_items = ", ".join(represent_python_value(item) for item in value)
trailing_comma = "," if len(value) == 1 else ""
return f"({rendered_items}{trailing_comma})"
case set() if not value:
return "set()"
case set():
sorted_items = sorted(value, key=lambda item: (type(item).__name__, repr(item)))
return "{" + ", ".join(represent_python_value(item) for item in sorted_items) + "}"
return repr(value)
Loading
Loading