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: 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, {}).items():
        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():
                _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:
        from protean.fields.resolved import convert_pydantic_errors

        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.
    """

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):
            # Only include non-None value objects
            dict_value = field_obj.as_dict(value)
            if dict_value:
                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

__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).
    """
    import copy

    if memo is None:
        memo = {}

    # Short-circuit if we've already been copied (prevents infinite loop)
    existing = 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