orchestrata_ultimate.core.domain.domain_entity

Reference › orchestrata_ultimate › core › domain › domain_entity

The domain layer's own identity-based base class. DomainEntity is a dataclass (declared with eq=False so that its custom equality is kept) with a UUID id and UTC created_at and updated_at timestamps. touch refreshes updated_at; equality and hashing use id alone; to_dict and from_dict serialise and rebuild the entity using the field type hints, with the same parser-registry helpers (_deserialize_list, _deserialize_dict, _deserialize_field) as base.entity. The module imports nothing from the project, and the listing shows no other module importing it.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

DomainEntity (dataclass)

Decorators: dataclass(eq=False) · Line 83

Base class for all domain entities.

Functions

Source

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

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

from __future__ import annotations

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[DomainEntity]) -> 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[DomainEntity]
) -> 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 DomainEntity:
    """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, DomainEntity):
                result[fname] = value.to_dict()
            elif isinstance(value, list):
                result[fname] = [
                    v.to_dict() if isinstance(v, DomainEntity) 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
        DomainEntity 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, DomainEntity)

        return cls(**kwargs)

    def __eq__(self, other: object) -> bool:
        """Identity equality: same id means equal."""
        if not isinstance(other, DomainEntity):
            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})"