from __future__ import annotations from contextlib import suppress from datetime import timezone from functools import partial from os.path import realpath from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, ForwardRef, Generic, Mapping, TypeVar, cast from uuid import NAMESPACE_DNS, uuid1, uuid3, uuid5 from typing_extensions import Literal, get_args from polyfactory.exceptions import MissingDependencyException from polyfactory.factories.base import BaseFactory, BuildContext from polyfactory.factories.base import BuildContext as BaseBuildContext from polyfactory.field_meta import Constraints, FieldMeta, Null from polyfactory.utils.deprecation import check_for_deprecated_parameters from polyfactory.utils.helpers import unwrap_new_type, unwrap_optional from polyfactory.utils.normalize_type import normalize_type from polyfactory.utils.predicates import is_annotated, is_optional, is_safe_subclass, is_union from polyfactory.utils.types import NoneType from polyfactory.value_generators.primitives import create_random_bytes try: import pydantic from pydantic import ( VERSION, AnyHttpUrl, AnyUrl, ByteSize, EmailStr, FutureDate, HttpUrl, IPvAnyAddress, IPvAnyInterface, IPvAnyNetwork, Json, NameEmail, NegativeFloat, NegativeInt, NonNegativeInt, NonPositiveFloat, PastDate, PaymentCardNumber, PositiveFloat, PositiveInt, SecretBytes, SecretStr, StrictBool, StrictBytes, StrictFloat, StrictInt, StrictStr, ) from pydantic.fields import FieldInfo except ImportError as e: msg = "pydantic is not installed" raise MissingDependencyException(msg) from e try: # pydantic v1 import pydantic as pydantic_v1 from pydantic import BaseModel as BaseModelV1 # Keep this import last to prevent warnings from pydantic if pydantic v2 # is installed. from pydantic.color import Color from pydantic.fields import ( # type: ignore[attr-defined] DeferredType, # pyright: ignore[attr-defined,reportAttributeAccessIssue] ModelField, # pyright: ignore[attr-defined,reportAttributeAccessIssue] Undefined, # pyright: ignore[attr-defined,reportAttributeAccessIssue] ) # prevent unbound variable warnings BaseModelV2 = BaseModelV1 UndefinedV2 = Undefined except ImportError: # pydantic v2 # v2 specific imports from pydantic import BaseModel as BaseModelV2 from pydantic_core import PydanticUndefined as UndefinedV2 from pydantic_core import to_json import pydantic.v1 as pydantic_v1 # type: ignore[no-redef] from pydantic.v1 import BaseModel as BaseModelV1 # type: ignore[assignment] from pydantic.v1.color import Color # type: ignore[assignment] from pydantic.v1.fields import DeferredType, ModelField, Undefined if TYPE_CHECKING: from collections import abc from random import Random from typing import Callable, Sequence from typing_extensions import NotRequired, TypeGuard from pydantic import BaseModel T = TypeVar("T", bound="BaseModel") _IS_PYDANTIC_V1 = VERSION.startswith("1") class PydanticBuildContext(BaseBuildContext): factory_use_construct: bool class PydanticConstraints(Constraints): """Metadata regarding a Pydantic type constraints, if any""" json: NotRequired[bool] class PydanticFieldMeta(FieldMeta): """Field meta subclass capable of handling pydantic ModelFields""" def __init__( self, *, name: str, annotation: type, random: Random | None = None, default: Any = ..., children: list[FieldMeta] | None = None, constraints: PydanticConstraints | None = None, examples: list[Any] | None = None, ) -> None: super().__init__( name=name, annotation=annotation, random=random, default=default, children=children, constraints=constraints, ) self.examples = examples @classmethod def from_field_info( cls, field_name: str, field_info: FieldInfo, use_alias: bool, random: Random | None = None, randomize_collection_length: bool | None = None, min_collection_length: int | None = None, max_collection_length: int | None = None, ) -> PydanticFieldMeta: """Create an instance from a pydantic field info. :param field_name: The name of the field. :param field_info: A pydantic FieldInfo instance. :param use_alias: Whether to use the field alias. :param random: A random.Random instance. :param randomize_collection_length: Whether to randomize collection length. :param min_collection_length: Minimum collection length. :param max_collection_length: Maximum collection length. :returns: A PydanticFieldMeta instance. """ check_for_deprecated_parameters( "2.11.0", parameters=( ("randomize_collection_length", randomize_collection_length), ("min_collection_length", min_collection_length), ("max_collection_length", max_collection_length), ("random", random), ), ) field_info = FieldInfo.merge_field_infos( field_info, FieldInfo.from_annotation(normalize_type(field_info.annotation)), alias=field_info.alias, ) if callable(field_info.default_factory): default_value = field_info.default_factory else: default_value = field_info.default if field_info.default is not UndefinedV2 else Null annotation = unwrap_new_type(field_info.annotation) children: list[FieldMeta,] | None = None name = field_info.alias if field_info.alias and use_alias else field_name constraints: PydanticConstraints # pydantic v2 does not always propagate metadata for Union types if is_union(annotation): constraints = {} children = [] # create a child for each of the possible union values for arg in get_args(annotation): # don't add the NoneType in an optional to the list of children if arg is NoneType: continue child_field_info = FieldInfo.from_annotation(arg) merged_field_info = FieldInfo.merge_field_infos(field_info, child_field_info) children.append( # recurse for each element of the union cls.from_field_info( # this is a fake field name, but it makes it possible to debug which type variant # is the source of an exception downstream field_name=field_name, field_info=merged_field_info, use_alias=use_alias, ), ) else: metadata, is_json = [], False for m in field_info.metadata: if not is_json and isinstance(m, Json): # type: ignore[misc] is_json = True elif m is not None: metadata.append(m) constraints = cast( "PydanticConstraints", cls.parse_constraints(metadata=metadata) if metadata else {}, ) if "url" in constraints: # pydantic uses a sentinel value for url constraints annotation = str if is_json: constraints["json"] = True result = super().from_type( annotation=annotation, children=children, constraints=cast("Constraints", {k: v for k, v in constraints.items() if v is not None}) or None, default=default_value, name=name, ) result.examples = field_info.examples return result @classmethod def from_model_field( # pragma: no cover cls, model_field: ModelField, # pyright: ignore[reportGeneralTypeIssues] use_alias: bool, randomize_collection_length: bool | None = None, min_collection_length: int | None = None, max_collection_length: int | None = None, random: Random | None = None, ) -> PydanticFieldMeta: """Create an instance from a pydantic model field. :param model_field: A pydantic ModelField. :param use_alias: Whether to use the field alias. :param randomize_collection_length: A boolean flag whether to randomize collections lengths :param min_collection_length: Minimum number of elements in randomized collection :param max_collection_length: Maximum number of elements in randomized collection :param random: An instance of random.Random. :returns: A PydanticFieldMeta instance. """ check_for_deprecated_parameters( "2.11.0", parameters=( ("randomize_collection_length", randomize_collection_length), ("min_collection_length", min_collection_length), ("max_collection_length", max_collection_length), ("random", random), ), ) if model_field.default is not Undefined: default_value = model_field.default elif callable(model_field.default_factory): default_value = model_field.default_factory() else: default_value = model_field.default if model_field.default is not Undefined else Null name = model_field.alias if model_field.alias and use_alias else model_field.name outer_type = unwrap_new_type(model_field.outer_type_) annotation = ( model_field.outer_type_ if isinstance(model_field.annotation, (DeferredType, ForwardRef)) else unwrap_new_type(model_field.annotation) ) # In pydantic v1, we need to check if the annotation is directly annotated to properly extract constraints # from the metadata, as v1 doesn't automatically propagate constraints like v2 does annotation_constraints: Constraints = {} if is_annotated(model_field.annotation): annotation_metadata = cls.get_constraints_metadata(model_field.annotation) annotation_constraints = cls.parse_constraints(annotation_metadata) if annotation_metadata else {} field_info_constraints = { "ge": getattr(outer_type, "ge", model_field.field_info.ge), "gt": getattr(outer_type, "gt", model_field.field_info.gt), "le": getattr(outer_type, "le", model_field.field_info.le), "lt": getattr(outer_type, "lt", model_field.field_info.lt), "min_length": ( getattr(outer_type, "min_length", model_field.field_info.min_length) or getattr(outer_type, "min_items", model_field.field_info.min_items) ), "max_length": ( getattr(outer_type, "max_length", model_field.field_info.max_length) or getattr(outer_type, "max_items", model_field.field_info.max_items) ), "pattern": getattr(outer_type, "regex", model_field.field_info.regex), "unique_items": getattr(outer_type, "unique_items", model_field.field_info.unique_items), "decimal_places": getattr(outer_type, "decimal_places", None), "max_digits": getattr(outer_type, "max_digits", None), "multiple_of": getattr(outer_type, "multiple_of", None), "upper_case": getattr(outer_type, "to_upper", None), "lower_case": getattr(outer_type, "to_lower", None), "item_type": getattr(outer_type, "item_type", None), } constraints = cast("Constraints", {**field_info_constraints, **annotation_constraints}) # pydantic v1 has constraints set for these values, but we generate them using faker if unwrap_optional(annotation) in ( AnyUrl, AnyHttpUrl, HttpUrl, pydantic_v1.AnyUrl, pydantic_v1.AnyHttpUrl, pydantic_v1.HttpUrl, pydantic_v1.KafkaDsn, pydantic_v1.PostgresDsn, pydantic_v1.RedisDsn, pydantic_v1.AmqpDsn, pydantic_v1.EmailStr, pydantic_v1.NameEmail, ): constraints = {} if model_field.field_info.const and ( default_value is None or isinstance(default_value, (int, bool, str, bytes)) ): annotation = Literal[default_value] children: list[FieldMeta] = [] # Refer #412. args = get_args(model_field.annotation) if is_optional(model_field.annotation) and len(args) == 2: # noqa: PLR2004 child_annotation = args[0] if args[0] is not NoneType else args[1] children.append(PydanticFieldMeta.from_type(child_annotation)) elif model_field.key_field or model_field.sub_fields: fields_to_iterate = ( ([model_field.key_field, *model_field.sub_fields]) if model_field.key_field is not None else model_field.sub_fields ) children.extend( PydanticFieldMeta.from_model_field( model_field=arg, use_alias=use_alias, ) for arg in fields_to_iterate ) examples = None return PydanticFieldMeta( name=name, annotation=annotation, # pyright: ignore[reportArgumentType] children=children or None, default=default_value, constraints=cast("PydanticConstraints", {k: v for k, v in constraints.items() if v is not None}) or None, examples=examples, ) if not _IS_PYDANTIC_V1: @classmethod def get_constraints_metadata(cls, annotation: Any) -> Sequence[Any]: metadata = [] for m in super().get_constraints_metadata(annotation): if isinstance(m, FieldInfo): metadata.extend(m.metadata) else: metadata.append(m) return metadata class ModelFactory(Generic[T], BaseFactory[T]): """Base factory for pydantic models""" __forward_ref_resolution_type_mapping__: ClassVar[Mapping[str, type]] = {} __is_base_factory__ = True __use_examples__: ClassVar[bool] = False # for backwards compatibility """ Flag indicating whether to use a random example, if provided (Pydantic >=V2) Example code:: class Payment(BaseModel): amount: int = Field(0) currency: str = Field(examples=['USD', 'EUR', 'INR']) class PaymentFactory(ModelFactory[Payment]): __use_examples__ = True >>> payment = PaymentFactory.build() >>> payment Payment(amount=120, currency="EUR") """ if not _IS_PYDANTIC_V1: __forward_references__: ClassVar[dict[str, Any]] = { # Resolve to str to avoid recursive issues "JsonValue": str, } __config_keys__ = ( *BaseFactory.__config_keys__, "__use_examples__", ) def __init_subclass__(cls, *args: Any, **kwargs: Any) -> None: super().__init_subclass__(*args, **kwargs) model = getattr(cls, "__model__", None) if model is None: return if _is_pydantic_v1_model(model) and hasattr(cls.__model__, "update_forward_refs"): with suppress(NameError): # pragma: no cover cls.__model__.update_forward_refs(**cls.__forward_ref_resolution_type_mapping__) if _is_pydantic_v2_model(model): model.model_rebuild() @classmethod def is_supported_type(cls, value: Any) -> TypeGuard[type[T]]: """Determine whether the given value is supported by the factory. :param value: An arbitrary value. :returns: A typeguard """ return _is_pydantic_v1_model(value) or _is_pydantic_v2_model(value) @classmethod def get_model_fields(cls) -> list["FieldMeta"]: """Retrieve a list of fields from the factory's model. :returns: A list of field MetaData instances. """ if "_fields_metadata" not in cls.__dict__: if _is_pydantic_v1_model(cls.__model__): cls._fields_metadata = [ PydanticFieldMeta.from_model_field( field, use_alias=not cls.__model__.__config__.allow_population_by_field_name, # type: ignore[attr-defined] ) for field in cls.__model__.__fields__.values() ] else: cls._fields_metadata = [ PydanticFieldMeta.from_field_info( field_info=field_info, field_name=field_name, use_alias=not cls.__model__.model_config.get( # pyright: ignore[reportGeneralTypeIssues] "populate_by_name", False, ), ) for field_name, field_info in cls.__model__.model_fields.items() # pyright: ignore[reportGeneralTypeIssues] ] return cls._fields_metadata @classmethod def get_constrained_field_value( cls, annotation: Any, field_meta: FieldMeta, field_build_parameters: Any | None = None, build_context: BuildContext | None = None, ) -> Any: constraints = cast("PydanticConstraints", field_meta.constraints) if constraints.pop("json", None): value = cls.get_field_value( field_meta, field_build_parameters=field_build_parameters, build_context=build_context ) return to_json(value) # pyright: ignore[reportPossiblyUnboundVariable] return super().get_constrained_field_value( annotation, field_meta, field_build_parameters=field_build_parameters, build_context=build_context ) @classmethod def get_field_value( cls, field_meta: FieldMeta, field_build_parameters: Any | None = None, build_context: BuildContext | None = None, ) -> Any: """Return a value from examples if exists, else random value. :param field_meta: FieldMeta instance. :param field_build_parameters: Any build parameters passed to the factory as kwarg values. :param build_context: BuildContext data for current build. :returns: An arbitrary value. """ result: Any field_meta = cast("PydanticFieldMeta", field_meta) if cls.__use_examples__ and field_meta.examples: result = cls.__random__.choice(field_meta.examples) else: result = super().get_field_value( field_meta=field_meta, field_build_parameters=field_build_parameters, build_context=build_context ) return result @classmethod def build( cls, factory_use_construct: bool = False, **kwargs: Any, ) -> T: """Build an instance of the factory's __model__ :param factory_use_construct: A boolean that determines whether validations will be made when instantiating the model. This is supported only for pydantic models. :param kwargs: Any kwargs. If field_meta names are set in kwargs, their values will be used. :returns: An instance of type T. """ if "_build_context" not in kwargs: kwargs["_build_context"] = PydanticBuildContext( seen_models=set(), factory_use_construct=factory_use_construct, ) processed_kwargs = cls.process_kwargs(**kwargs) return cls._create_model(kwargs["_build_context"], **processed_kwargs) @classmethod def _get_build_context(cls, build_context: BaseBuildContext | PydanticBuildContext | None) -> PydanticBuildContext: """Return a PydanticBuildContext instance. If build_context is None, return a new PydanticBuildContext. :returns: PydanticBuildContext """ build_context = cast("PydanticBuildContext", super()._get_build_context(build_context)) if build_context.get("factory_use_construct") is None: build_context["factory_use_construct"] = False return build_context @classmethod def _create_model(cls, _build_context: PydanticBuildContext, **kwargs: Any) -> T: """Create an instance of the factory's __model__ :param _build_context: BuildContext instance. :param kwargs: Model kwargs. :returns: An instance of type T. """ if _build_context.get("factory_use_construct"): if _is_pydantic_v1_model(cls.__model__): return cls.__model__.construct(**kwargs) # type: ignore[return-value] return cls.__model__.model_construct(**kwargs) return cls.__model__(**kwargs) @classmethod def coverage(cls, factory_use_construct: bool = False, **kwargs: Any) -> abc.Iterator[T]: """Build a batch of the factory's Meta.model will full coverage of the sub-types of the model. :param kwargs: Any kwargs. If field_meta names are set in kwargs, their values will be used. :returns: A iterator of instances of type T. """ if "_build_context" not in kwargs: kwargs["_build_context"] = PydanticBuildContext( seen_models=set(), factory_use_construct=factory_use_construct ) for data in cls.process_kwargs_coverage(**kwargs): yield cls._create_model(_build_context=kwargs["_build_context"], **data) @classmethod def is_custom_root_field(cls, field_meta: FieldMeta) -> bool: """Determine whether the field is a custom root field. :param field_meta: FieldMeta instance. :returns: A boolean determining whether the field is a custom root. """ return field_meta.name == "__root__" @classmethod def should_set_field_value(cls, field_meta: FieldMeta, **kwargs: Any) -> bool: """Determine whether to set a value for a given field_name. This is an override of BaseFactory.should_set_field_value. :param field_meta: FieldMeta instance. :param kwargs: Any kwargs passed to the factory. :returns: A boolean determining whether a value should be set for the given field_meta. """ return field_meta.name not in kwargs and ( not field_meta.name.startswith("_") or cls.is_custom_root_field(field_meta) ) @classmethod def get_provider_map(cls) -> dict[Any, Callable[[], Any]]: mapping: dict[Any, Callable[[], Any]] = { ByteSize: cls.__faker__.pyint, PositiveInt: cls.__faker__.pyint, NegativeFloat: lambda: cls.__random__.uniform(-100, -1), NegativeInt: lambda: cls.__faker__.pyint() * -1, PositiveFloat: cls.__faker__.pyint, NonPositiveFloat: lambda: cls.__random__.uniform(-100, 0), NonNegativeInt: cls.__faker__.pyint, StrictInt: cls.__faker__.pyint, StrictBool: cls.__faker__.pybool, StrictBytes: lambda: create_random_bytes(cls.__random__), StrictFloat: cls.__faker__.pyfloat, StrictStr: cls.__faker__.pystr, EmailStr: cls.__faker__.free_email, NameEmail: cls.__faker__.free_email, Json: cls.__faker__.json, PaymentCardNumber: cls.__faker__.credit_card_number, AnyUrl: cls.__faker__.url, AnyHttpUrl: cls.__faker__.url, HttpUrl: cls.__faker__.url, SecretBytes: lambda: create_random_bytes(cls.__random__), SecretStr: cls.__faker__.pystr, IPvAnyAddress: cls.__faker__.ipv4, IPvAnyInterface: cls.__faker__.ipv4, IPvAnyNetwork: lambda: cls.__faker__.ipv4(network=True), PastDate: cls.__faker__.past_date, FutureDate: cls.__faker__.future_date, } # v1 only values mapping.update( { pydantic_v1.AnyUrl: cls.__faker__.url, pydantic_v1.AnyHttpUrl: cls.__faker__.url, pydantic_v1.HttpUrl: cls.__faker__.url, pydantic_v1.PyObject: lambda: "decimal.Decimal", pydantic_v1.AmqpDsn: lambda: "amqps://example.com", pydantic_v1.KafkaDsn: lambda: "kafka://localhost:9092", pydantic_v1.PostgresDsn: lambda: "postgresql://user@localhost", pydantic_v1.RedisDsn: lambda: "redis://localhost:6379/0", pydantic_v1.FilePath: lambda: Path(realpath(__file__)), pydantic_v1.DirectoryPath: lambda: Path(realpath(__file__)).parent, pydantic_v1.UUID1: uuid1, pydantic_v1.UUID3: lambda: uuid3(NAMESPACE_DNS, cls.__faker__.pystr()), pydantic_v1.UUID4: cls.__faker__.uuid4, pydantic_v1.UUID5: lambda: uuid5(NAMESPACE_DNS, cls.__faker__.pystr()), Color: cls.__faker__.hex_color, # pyright: ignore[reportGeneralTypeIssues] pydantic_v1.EmailStr: cls.__faker__.free_email, pydantic_v1.NameEmail: cls.__faker__.free_email, }, ) if not _IS_PYDANTIC_V1: mapping.update( { # pydantic v2 specific types pydantic.PastDatetime: cls.__faker__.past_datetime, pydantic.FutureDatetime: cls.__faker__.future_datetime, pydantic.AwareDatetime: partial(cls.__faker__.date_time, timezone.utc), pydantic.NaiveDatetime: cls.__faker__.date_time, pydantic.networks.AmqpDsn: lambda: "amqps://example.com", pydantic.networks.KafkaDsn: lambda: "kafka://localhost:9092", pydantic.networks.PostgresDsn: lambda: "postgresql://user@localhost", pydantic.networks.RedisDsn: lambda: "redis://localhost:6379/0", pydantic.networks.MongoDsn: lambda: "mongodb://mongodb0.example.com:27017", pydantic.networks.MariaDBDsn: lambda: "mariadb://example.com:3306", pydantic.networks.CockroachDsn: lambda: "cockroachdb://example.com:5432", pydantic.networks.MySQLDsn: lambda: "mysql://example.com:5432", }, ) mapping.update(super().get_provider_map()) return mapping def _is_pydantic_v1_model(model: Any) -> TypeGuard[BaseModelV1]: return is_safe_subclass(model, BaseModelV1) def _is_pydantic_v2_model(model: Any) -> TypeGuard[BaseModelV2]: # pyright: ignore[reportInvalidTypeForm] return not _IS_PYDANTIC_V1 and is_safe_subclass(model, BaseModelV2)