ValueObject field

Field descriptor for embedding value objects within aggregates or entities.

Bases: Field

Represents a field that holds a value object.

This field is used to embed a value object within an entity. It provides functionality to handle the value object's fields and their values.

PARAMETER	DESCRIPTION
value_object_cls	The class of the value object to be embedded. TYPE: class

ATTRIBUTE	DESCRIPTION
embedded_fields	A dictionary that holds the embedded fields of the value object. TYPE: dict

Source code in src/protean/fields/embedded.py

def __init__(
    self,
    value_object_cls: "type[BaseValueObject] | str",
    *args: Any,
    **kwargs: Any,
) -> None:
    super().__init__(*args, **kwargs)

    if not isinstance(value_object_cls, str):
        # Validate the class being passed is a subclass of BaseValueObject
        self._validate_value_object_cls(value_object_cls)

    self._value_object_cls: "type[BaseValueObject] | str" = value_object_cls

    self._embedded_fields: dict[str, _ShadowField] = {}

embedded_fields cached property

embedded_fields: dict[str, _ShadowField]

Property to retrieve embedded fields

get_shadow_fields

get_shadow_fields() -> list[
    tuple[str | None, _ShadowField]
]

Return shadow field Primarily used during Entity initialization to register shadow field

Source code in src/protean/fields/embedded.py

def get_shadow_fields(self) -> list[tuple[str | None, _ShadowField]]:
    """Return shadow field
    Primarily used during Entity initialization to register shadow field"""
    shadow_fields: list[tuple[str | None, _ShadowField]] = []
    for field in self.embedded_fields.values():
        shadow_fields.append((field.attribute_name, field))
    return shadow_fields

as_dict

as_dict(value: Any) -> Any

Return JSON-compatible value of self

Source code in src/protean/fields/embedded.py

def as_dict(self, value: Any) -> Any:
    """Return JSON-compatible value of self"""
    return (
        {
            field_name: shadow_field_obj.field_obj.as_dict(
                getattr(value, field_name, None)
            )
            for field_name, shadow_field_obj in self.embedded_fields.items()
        }
        # Presence is identity, not truthiness (#1078): an all-default VO
        # (e.g. all zeros) is falsy per ``__bool__`` but was explicitly set,
        # so it must serialize to its dict rather than to ``None``.
        if value is not None
        else None
    )

__set__

__set__(instance: Any, value: Any) -> None

Override __set__ to coordinate between value object and its embedded fields

Source code in src/protean/fields/embedded.py

def __set__(self, instance: Any, value: Any) -> None:
    """Override `__set__` to coordinate between value object and its embedded fields"""
    value = self._load(value)

    # Identity, not truthiness — see ``as_dict`` (#1078). ``if value`` would
    # reset an all-default (falsy) VO, erasing "all-defaults" vs "never set".
    if value is not None:
        # Check if the reference object has been saved. Otherwise, throw ValueError
        self._set_own_value(instance, value)
        self._set_embedded_values(instance, value)
    else:
        self._reset_values(instance)

    # Mark Entity as Dirty
    if hasattr(instance, "state_"):
        instance.state_.mark_changed()

---

ValueObjectFromEntity field

Field descriptor that auto-generates a value object from an entity class. Convenience wrapper over value_object_from_entity() for inline use in commands and events.

Bases: ValueObject

Field descriptor that auto-generates a Value Object from an Entity class.

Thin convenience wrapper over value_object_from_entity() -- instead of creating the VO class yourself, this descriptor derives it from the entity at class-body evaluation time::

class PlaceOrder(BaseCommand):
    items: List(content_type=ValueObjectFromEntity(OrderItem))

Source code in src/protean/fields/embedded.py

def __init__(self, entity_cls: type, **kwargs: Any) -> None:
    from protean.core.value_object import value_object_from_entity  # noqa: PLC0415

    vo_cls = value_object_from_entity(entity_cls)
    super().__init__(value_object_cls=vo_cls, **kwargs)

---

value_object_from_entity

Utility function that creates a BaseValueObject subclass mirroring an entity's fields. See the Value Objects guide for usage patterns.

Create a BaseValueObject subclass whose fields mirror entity_cls.

This eliminates the boilerplate of manually duplicating an entity's fields into a value object for use in commands and events.

PARAMETER	DESCRIPTION
entity_cls	The entity (or aggregate) class to project. TYPE: type[Any]
name	Override the generated class name (default: {EntityName}ValueObject). TYPE: str | None DEFAULT: None
exclude	Field names to omit from the generated value object. TYPE: set[str] | None DEFAULT: None

RETURNS	DESCRIPTION
type[BaseValueObject]	A new BaseValueObject subclass.

Source code in src/protean/core/value_object.py

def value_object_from_entity(
    entity_cls: type[Any],
    name: str | None = None,
    exclude: set[str] | None = None,
) -> type[BaseValueObject]:
    """Create a ``BaseValueObject`` subclass whose fields mirror *entity_cls*.

    This eliminates the boilerplate of manually duplicating an entity's fields
    into a value object for use in commands and events.

    Args:
        entity_cls: The entity (or aggregate) class to project.
        name: Override the generated class name (default: ``{EntityName}ValueObject``).
        exclude: Field names to omit from the generated value object.

    Returns:
        A new ``BaseValueObject`` subclass.
    """
    exclude = exclude or set()
    vo_name = name or f"{entity_cls.__name__}ValueObject"

    annotations: dict[str, Any] = {}
    namespace: dict[str, Any] = {}
    # ValueObjectList is lazily imported above; use Any in the annotation
    # to avoid referencing it at module-import time.
    association_descriptors: dict[str, Any] = {}
    model_field_info = getattr(entity_cls, "model_fields", {})

    for key, value in get_fields(entity_cls).items():
        if key in exclude:
            continue
        if isinstance(value, Reference):
            continue
        if key.startswith("_"):
            continue

        if isinstance(value, HasOne):
            # Recursively convert associated entity to VO. ``to_cls`` is
            # ``str | type`` (lazy string ref or resolved class), but by the
            # time an entity is projected into a VO the target is a resolved
            # class.
            child_vo_cls = value_object_from_entity(cast("type[Any]", value.to_cls))
            vo_descriptor = ValueObjectField(value_object_cls=child_vo_cls)
            annotations[key] = Optional[child_vo_cls]
            namespace[key] = None
            association_descriptors[key] = vo_descriptor

        elif isinstance(value, HasMany):
            # Recursively convert to list of VOs. See ``to_cls`` narrowing note
            # above.
            child_vo_cls = value_object_from_entity(cast("type[Any]", value.to_cls))
            vo_descriptor = ValueObjectField(value_object_cls=child_vo_cls)
            # ``ValueObjectList.__init__`` has an untyped ``content_type``
            # parameter that defaults to ``str``, so pyright infers ``type[str]``.
            # A ValueObject descriptor is a documented, runtime-accepted value
            # for lists of value objects; the constructor validates it explicitly.
            list_descriptor = ValueObjectList(content_type=vo_descriptor)  # pyright: ignore[reportArgumentType]
            # ``child_vo_cls`` is a runtime-built VO class; mypy cannot model a
            # generic subscripted by a local variable (treats it as a type-alias
            # attempt). The runtime construction is valid.
            annotations[key] = list[child_vo_cls]  # type: ignore[valid-type]
            namespace[key] = PydanticField(default_factory=list)
            association_descriptors[key] = list_descriptor

        elif isinstance(value, ValueObjectField):
            vo_cls = value.value_object_cls
            # Preserve the required flag from the original descriptor
            if getattr(value, "required", False):
                annotations[key] = vo_cls
            else:
                annotations[key] = Optional[vo_cls]
                namespace[key] = None
            association_descriptors[key] = value

        elif isinstance(value, ResolvedField):
            finfo = model_field_info.get(key)
            if finfo:
                annotations[key] = finfo.annotation
                if finfo.default is not PydanticUndefined:
                    namespace[key] = finfo.default
                elif finfo.default_factory is not None:
                    namespace[key] = PydanticField(
                        default_factory=finfo.default_factory
                    )

    # Make identifier/unique fields optional (they are identity concerns,
    # not value concerns).
    container_fields = get_fields(entity_cls)
    for key in list(annotations.keys()):
        field_obj = container_fields.get(key)
        if isinstance(field_obj, ResolvedField) and (
            field_obj.identifier or field_obj.unique
        ):
            annotations[key] = annotations[key] | None
            namespace[key] = None

    ns = {"__annotations__": annotations, **namespace}
    value_object_cls = type(vo_name, (BaseValueObject,), ns)

    # Inject association descriptors into __container_fields__
    cf = getattr(value_object_cls, _FIELDS, {})
    cf.update(association_descriptors)
    setattr(value_object_cls, _FIELDS, cf)

    return value_object_cls