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, *args, **kwargs):
    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 = value_object_cls

    self._embedded_fields = {}

embedded_fields cached property

embedded_fields

Property to retrieve embedded fields

get_shadow_fields

get_shadow_fields()

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):
    """Return shadow field
    Primarily used during Entity initialization to register shadow field"""
    shadow_fields = []
    for field in self.embedded_fields.values():
        shadow_fields.append((field.attribute_name, field))
    return shadow_fields

as_dict

as_dict(value)

Return JSON-compatible value of self

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

def as_dict(self, value):
    """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()
        }
        if value
        else None
    )

__set__

__set__(instance, value)

Override __set__ to coordinate between value object and its embedded fields

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

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

    if value:
        # 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

    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
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,
    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.
    """
    from protean.fields.basic import ValueObjectList

    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
            child_vo_cls = value_object_from_entity(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
            child_vo_cls = value_object_from_entity(value.to_cls)
            vo_descriptor = ValueObjectField(value_object_cls=child_vo_cls)
            list_descriptor = ValueObjectList(content_type=vo_descriptor)
            annotations[key] = list[child_vo_cls]
            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