BaseEntity

Base class for entities, domain objects with unique identity that live within an aggregate. Entities are always accessed through their parent aggregate and cannot exist independently.

See Entities guide for practical usage and Entities concept for design rationale.

Bases: Element, BaseModel, OptionsMixin

Base class for entities -- domain objects with unique identity that live within an aggregate.

Entities are mutable with identity-based equality: two entities are equal if they share the same identifier, regardless of attribute values. Field mutations are validated via Pydantic's validate_assignment and trigger pre/post invariant checks automatically.

Fields are declared using standard Python type annotations with optional Field constraints. An identity field is auto-injected unless one is explicitly declared with identifier=True.

Entities track their lifecycle state (new, persisted, changed, destroyed) via the state_ property, and can raise events that are registered on the aggregate root.

Meta Options

Option	Type	Description
part_of	type	The aggregate class this entity belongs to. Required.
provider	str	The persistence provider name (default: "default").
schema_name	str	The storage table/collection name.
auto_add_id_field	bool	Whether to auto-inject an id field (default: True).

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

def __init__(self, *args: Any, **kwargs: Any) -> None:
    # Pop internal kwargs that should not reach Pydantic
    owner = kwargs.pop("_owner", None)
    root = kwargs.pop("_root", None)
    # _version is a PrivateAttr; accept it in kwargs for backward
    # compatibility (e.g. repository hydration) but set it after init.
    _version_value = kwargs.pop("_version", None)

    # Pop association/VO descriptor kwargs and shadow field kwargs before
    # Pydantic init.  Shadow fields (e.g. order_id, billing_address_street)
    # are set dynamically during domain resolution and must not be passed
    # to Pydantic's __init__ which rejects them with extra="forbid".
    descriptor_kwargs: dict[str, Any] = {}
    shadow_kwargs: dict[str, Any] = {}

    # Build the set of known shadow field names from descriptors in
    # __container_fields__.  This prevents silently swallowing truly
    # unknown kwargs that Pydantic should reject.
    _shadow_field_names: set[str] = set()
    for fobj in getattr(type(self), _FIELDS, {}).values():
        if isinstance(fobj, Reference):
            attr_name = fobj.get_attribute_name()
            if attr_name:
                _shadow_field_names.add(attr_name)
        elif isinstance(fobj, ValueObject):
            for sf in fobj.embedded_fields.values():
                if sf.attribute_name is not None:
                    _shadow_field_names.add(sf.attribute_name)

    for name in list(kwargs):
        if name in _shadow_field_names:
            shadow_kwargs[name] = kwargs.pop(name)
        elif self._get_class_descriptor(type(self), name) is not None:
            descriptor_kwargs[name] = kwargs.pop(name)

    # Support template dict pattern: Entity({"key": "val"}, key2="val2")
    # Keyword args take precedence over template dict values.
    if args:
        merged: dict[str, Any] = {}
        for template in args:
            if not isinstance(template, dict):
                raise AssertionError(
                    f"Positional argument {template} passed must be a dict. "
                    f"This argument serves as a template for loading common "
                    f"values.",
                )
            # Also separate descriptor and shadow kwargs from template dicts
            for tname in list(template):
                if tname in _shadow_field_names:
                    shadow_kwargs[tname] = template.pop(tname)
                elif self._get_class_descriptor(type(self), tname) is not None:
                    descriptor_kwargs[tname] = template.pop(tname)
            merged.update(template)
        merged.update(kwargs)
        kwargs = merged

    # Collect all validation errors (Pydantic + required descriptors) before raising
    collected_errors: dict[str, list[str]] = {}

    # Push init context onto the thread-local stack so that
    # model_post_init (called by Pydantic inside super().__init__())
    # can retrieve descriptor/shadow kwargs.  Pydantic wipes __dict__
    # during validation, so we cannot stash data on the instance.
    stack: list[dict[str, Any]] = getattr(_init_context, "stack", [])
    stack.append(
        {
            "descriptor_kwargs": descriptor_kwargs,
            "shadow_kwargs": shadow_kwargs,
            "owner": owner,
            "root": root,
            "_version": _version_value,
        }
    )
    _init_context.stack = stack

    try:
        super().__init__(**kwargs)
    except PydanticValidationError as e:
        collected_errors.update(convert_pydantic_errors(e))

    # Check required descriptor fields (ValueObject, Reference, etc.)
    for field_name, field_obj in getattr(type(self), _FIELDS, {}).items():
        if (
            isinstance(field_obj, (ValueObject, Reference))
            and getattr(field_obj, "required", False)
            and field_name not in descriptor_kwargs
        ):
            # Check if shadow fields are present (e.g. region_id for region Reference)
            has_shadow = False
            if isinstance(field_obj, ValueObject):
                has_shadow = any(
                    sf.attribute_name in shadow_kwargs
                    for sf in field_obj.embedded_fields.values()
                )
            elif isinstance(field_obj, Reference):
                shadow_name = field_obj.get_attribute_name()
                has_shadow = shadow_name in shadow_kwargs

            if not has_shadow:
                collected_errors.setdefault(field_name, []).append("is required")

    if collected_errors:
        raise ValidationError(collected_errors)

state_ property writable

state_: _EntityState

Access entity lifecycle state.

__track_id_field classmethod

__track_id_field() -> None

Find the field marked identifier=True and record its name.

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

@classmethod
def __track_id_field(cls) -> None:
    """Find the field marked ``identifier=True`` and record its name."""
    id_fields = [
        field
        for _, field in getattr(cls, _FIELDS, {}).items()
        if getattr(field, "identifier", False)
    ]

    if len(id_fields) > 1:
        raise NotSupportedError(
            {
                "_entity": [
                    f"Multiple identifier fields found in entity {cls.__name__}. "
                    "Only one identifier field is allowed."
                ]
            }
        )
    elif len(id_fields) == 1:
        setattr(cls, _ID_FIELD_NAME, id_fields[0].field_name)

defaults

defaults() -> None

Placeholder for defaults.

Override in subclass when an attribute's default depends on other attribute values.

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

def defaults(self) -> None:
    """Placeholder for defaults.

    Override in subclass when an attribute's default depends on other attribute values.
    """

can_transition_to

can_transition_to(
    field_name: str, target_value: Any
) -> bool

Check whether a status field can transition to the given value.

Returns True if the transition is valid, False otherwise. Always returns True for non-status fields or status fields without a transitions map.

Example::

if order.can_transition_to("status", OrderStatus.SHIPPED):
    order.ship(tracking_number="ABC123")

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

def can_transition_to(self, field_name: str, target_value: Any) -> bool:
    """Check whether a status field can transition to the given value.

    Returns ``True`` if the transition is valid, ``False`` otherwise.
    Always returns ``True`` for non-status fields or status fields
    without a transitions map.

    Example::

        if order.can_transition_to("status", OrderStatus.SHIPPED):
            order.ship(tracking_number="ABC123")
    """

    fields_dict = getattr(self.__class__, _FIELDS, {})
    field_obj = fields_dict.get(field_name)
    if field_obj is None or not isinstance(field_obj, ResolvedField):
        return True
    transitions = field_obj.transitions
    if not transitions:
        return True

    current_value = getattr(self, field_name, None)
    target = target_value.value if isinstance(target_value, Enum) else target_value
    current = (
        current_value.value if isinstance(current_value, Enum) else current_value
    )

    if current is None:
        return True

    if current not in transitions:
        return False

    return target in transitions[current]

raise_

raise_(event: Any) -> None

Raise an event in the aggregate cluster.

The event is always registered on the aggregate root, irrespective of where it is raised in the entity cluster.

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

def raise_(self, event: Any) -> None:
    """Raise an event in the aggregate cluster.

    The event is always registered on the aggregate root, irrespective
    of where it is raised in the entity cluster.
    """
    if event.meta_.part_of != self._root.__class__:
        raise ConfigurationError(
            f"Event `{event.__class__.__name__}` is not associated with"
            f" aggregate `{self._root.__class__.__name__}`"
        )

    # Delegate to the root aggregate's raise_ (BaseAggregate overrides this)
    self._root.raise_(event)

to_dict

to_dict() -> dict[str, Any]

Return entity data as a dictionary.

Reference fields are skipped (they are navigation, not data). ValueObject fields are included only when non-None.

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

def to_dict(self) -> dict[str, Any]:
    """Return entity data as a dictionary.

    Reference fields are skipped (they are navigation, not data).
    ValueObject fields are included only when non-None.
    """
    result: dict[str, Any] = {}
    for fname, field_obj in getattr(self, _FIELDS, {}).items():
        if isinstance(field_obj, Reference):
            continue

        value = getattr(self, fname, None)

        if isinstance(field_obj, ValueObject):
            # Present, not truthy: ``as_dict`` returns ``None`` only
            # when the VO is ``None``, so an all-default VO is still included.
            dict_value = field_obj.as_dict(value)
            if dict_value is not None:
                result[fname] = dict_value
        elif isinstance(field_obj, Association):
            # HasOne/HasMany: delegate to descriptor's as_dict
            result[fname] = field_obj.as_dict(value)
        else:
            result[fname] = field_obj.as_dict(value)
    return result

from_value_object classmethod

from_value_object(vo: BaseValueObject) -> Self

Construct an entity instance from a value object.

This is the inverse of value_object_from_entity() -- it converts a VO payload (typically carried in a command or event) back into an entity instance::

items = [OrderItem.from_value_object(item) for item in command.items]

None values on identifier/unique fields are stripped so that auto-generated identity fields receive their default rather than failing validation. Explicit None on other fields is preserved.

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

@classmethod
def from_value_object(cls, vo: "BaseValueObject") -> Self:
    """Construct an entity instance from a value object.

    This is the inverse of ``value_object_from_entity()`` -- it converts
    a VO payload (typically carried in a command or event) back into an
    entity instance::

        items = [OrderItem.from_value_object(item) for item in command.items]

    ``None`` values on identifier/unique fields are stripped so that
    auto-generated identity fields receive their default rather than
    failing validation.  Explicit ``None`` on other fields is preserved.
    """
    data = vo.to_dict()

    # Strip None only for identifier/unique fields so auto-generated
    # identity defaults kick in, while preserving intentional Nones.
    id_or_unique = set()
    for fname, field_obj in getattr(cls, _FIELDS, {}).items():
        if isinstance(field_obj, ResolvedField) and (
            field_obj.identifier or field_obj.unique
        ):
            id_or_unique.add(fname)

    return cls(
        **{k: v for k, v in data.items() if not (v is None and k in id_or_unique)}
    )

__deepcopy__

__deepcopy__(
    memo: dict[int, Any] | None = None,
) -> BaseEntity

Deep copy that handles circular _root/_owner references.

Pydantic's default deepcopy recurses infinitely when pydantic_private contains back-references to the entity itself (e.g. _root and _owner on aggregate roots).

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

def __deepcopy__(self, memo: dict[int, Any] | None = None) -> "BaseEntity":
    """Deep copy that handles circular _root/_owner references.

    Pydantic's default __deepcopy__ recurses infinitely when
    __pydantic_private__ contains back-references to the entity
    itself (e.g. _root and _owner on aggregate roots).
    """
    if memo is None:
        memo = {}

    # Short-circuit if we've already been copied (prevents infinite loop)
    existing: BaseEntity | None = memo.get(id(self))
    if existing is not None:
        return existing

    cls = type(self)
    new_obj = cls.__new__(cls)
    memo[id(self)] = new_obj

    # Deep-copy __dict__ (Pydantic model fields and extras)
    object.__setattr__(new_obj, "__dict__", copy.deepcopy(self.__dict__, memo))
    object.__setattr__(
        new_obj,
        "__pydantic_extra__",
        copy.deepcopy(self.__pydantic_extra__, memo),
    )
    object.__setattr__(
        new_obj,
        "__pydantic_fields_set__",
        copy.copy(self.__pydantic_fields_set__),
    )

    # Deep-copy __pydantic_private__, with memo to break cycles
    private = getattr(self, "__pydantic_private__", None)
    if private is None:
        object.__setattr__(new_obj, "__pydantic_private__", None)
    else:
        object.__setattr__(
            new_obj,
            "__pydantic_private__",
            copy.deepcopy(private, memo),
        )

    return new_obj