orchestrata_ultimate.core.base.entity

Reference › orchestrata_ultimate › core › base › entity

The identity-based domain object that generated and ported classes inherit from, in the base package. BaseEntity is a dataclass with a UUID string id and two UTC timestamps, created_at and updated_at. touch refreshes updated_at. Equality and hashing use the id alone, so two instances with the same id are the same entity and can be used in sets and as dictionary keys. repr shows the class name and id.

to_dict serialises an entity recursively. from_dict rebuilds one using the field type hints: a registry of type parsers (currently datetime.fromisoformat) handles directly typed values, and _deserialize_field, _deserialize_list and _deserialize_dict handle list[T], dict[K, V] and nested entity subclasses; unrecognised types pass through unchanged. DomainEntity is declared here as an empty subclass of BaseEntity, an alias for it; the domain package has its own separate DomainEntity. The persistence contract for entities is repository.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

BaseEntity (dataclass)

Decorators: dataclass(eq=False) · Line 81

Base class for all domain entities.

DomainEntity

Bases: BaseEntity · Line 174

Domain entity alias for BaseEntity.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/base/entity.py

"""BaseEntity — identity-based domain object with UUID and timestamps."""

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, ClassVar, Self

# Type-parser registry for from_dict deserialization
_TYPE_PARSERS: dict[type, Any] = {
    datetime: datetime.fromisoformat,
}


def _deserialize_list(inner: Any, value: list, entity_base: type[BaseEntity]) -> list:
    """Deserialize a ``list[inner]`` value using the parser registry.

    Each element is parsed when *inner* has a registered parser, reconstructed
    via ``from_dict`` when *inner* is an *entity_base* subclass, and passed
    through otherwise.
    """
    if inner in _TYPE_PARSERS:
        parser = _TYPE_PARSERS[inner]
        return [parser(v) if isinstance(v, str) else v for v in value]
    if isinstance(inner, type) and issubclass(inner, entity_base):
        return [inner.from_dict(v) if isinstance(v, dict) else v for v in value]
    return value


def _deserialize_dict(val_type: Any, value: dict) -> dict:
    """Deserialize a ``dict[K, val_type]`` value using the parser registry."""
    parser = _TYPE_PARSERS[val_type]
    return {k: parser(v) if isinstance(v, str) else v for k, v in value.items()}


def _deserialize_field(
    ftype: Any, origin: Any, value: Any, entity_base: type[BaseEntity]
) -> Any:
    """Deserialize a single field *value* according to its type hint.

    Dispatches on the kind of *ftype*: a directly-registered parser type,
    a ``list[T]``, a ``dict[K, V]``, a nested *entity_base* subclass, or an
    unrecognised type (passed through unchanged).
    """
    import typing

    # Direct type with parser (datetime, etc.)
    if ftype in _TYPE_PARSERS and isinstance(value, str):
        return _TYPE_PARSERS[ftype](value)

    # list[T] where T has a parser or is an entity subclass
    if origin is list:
        args = typing.get_args(ftype)
        if args and isinstance(value, list):
            return _deserialize_list(args[0], value, entity_base)
        return value

    # dict[K, V] where V has a parser
    if origin is dict:
        args = typing.get_args(ftype)
        if (
            args
            and len(args) == 2
            and args[1] in _TYPE_PARSERS
            and isinstance(value, dict)
        ):
            return _deserialize_dict(args[1], value)
        return value

    # Nested entity subclass
    if (
        isinstance(ftype, type)
        and issubclass(ftype, entity_base)
        and isinstance(value, dict)
    ):
        return ftype.from_dict(value)

    return value


@dataclass(eq=False)
class BaseEntity:
    """Base class for all domain entities.

    Provides UUID identity, timestamps, serialization, and
    identity-based equality (same id = equal).

    Attributes:
        id: Unique identifier (uuid4 string).
        created_at: Creation timestamp (UTC).
        updated_at: Last modification timestamp (UTC).
    """

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    # Subclass type-hint cache for from_dict
    _field_types: ClassVar[dict[str, type]] = {}

    def touch(self) -> None:
        """Update the updated_at timestamp to now."""
        self.updated_at = datetime.now(UTC)

    def to_dict(self) -> dict[str, Any]:
        """Serialize entity to a dictionary.

        Returns:
            Dict with all dataclass fields serialized.
        """
        result: dict[str, Any] = {}
        for fname, ftype in self.__dataclass_fields__.items():
            value = getattr(self, fname)
            if isinstance(value, datetime):
                result[fname] = value.isoformat()
            elif isinstance(value, BaseEntity):
                result[fname] = value.to_dict()
            elif isinstance(value, list):
                result[fname] = [
                    v.to_dict() if isinstance(v, BaseEntity) else v for v in value
                ]
            else:
                result[fname] = value
        return result

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Self:
        """Deserialize entity from a dictionary.

        Handles datetime parsing and single-depth nested
        BaseEntity reconstruction via type-hint introspection.

        Args:
            data: Dictionary of field values.

        Returns:
            New entity instance.

        Raises:
            TypeError: For unsupported Union types.
        """
        import typing

        hints = typing.get_type_hints(cls)
        kwargs: dict[str, Any] = {}

        for fname, ftype in hints.items():
            if fname.startswith("_"):
                continue
            if fname not in data:
                continue

            value = data[fname]
            origin = typing.get_origin(ftype)
            kwargs[fname] = _deserialize_field(ftype, origin, value, BaseEntity)

        return cls(**kwargs)

    def __eq__(self, other: object) -> bool:
        """Identity equality: same id means equal."""
        if not isinstance(other, BaseEntity):
            return NotImplemented
        return self.id == other.id

    def __hash__(self) -> int:
        """Hash based on id for use in sets and dict keys."""
        return hash(self.id)

    def __repr__(self) -> str:
        """Short repr with class name and id."""
        return f"{self.__class__.__name__}(id={self.id!r})"


# Class alias for orchestrata-ultimate naming
class DomainEntity(BaseEntity):
    """Domain entity alias for BaseEntity."""

    pass