spakky-domain

DDD 빌딩 블록 — Aggregate Root, Entity, Value Object, Domain Event, CQRS

모델

spakky.domain.models.aggregate_root

Aggregate root model for domain-driven design.

This module provides AbstractAggregateRoot for representing DDD aggregate roots that manage domain events and maintain consistency boundaries.

AggregateRootT = TypeVar('AggregateRootT', bound=AbstractAggregateRoot) module-attribute

Type variable for aggregate root types (invariant for repositories).

AggregateRootT_co = TypeVar('AggregateRootT_co', bound=AbstractAggregateRoot, covariant=True) module-attribute

Type variable for aggregate root types (covariant for read-only operations).

AggregateRootT_contra = TypeVar('AggregateRootT_contra', bound=AbstractAggregateRoot, contravariant=True) module-attribute

Type variable for aggregate root types (contravariant for input parameters).

AbstractAggregateRoot

Bases: AbstractEntity[EquatableT], ABC

Base class for DDD aggregate roots.

Aggregate roots are entities that serve as entry points to aggregates, maintaining consistency boundaries and managing domain events.

events property

Get copy of all domain events raised by this aggregate.

Returns:

Type	Description
Sequence[AbstractDomainEvent]	Sequence of domain events.

add_event(event)

Add a domain event to this aggregate.

Parameters:

Name	Type	Description	Default
event	AbstractDomainEvent	The domain event to add.	required

Source code in core/spakky-domain/src/spakky/domain/models/aggregate_root.py

def add_event(self, event: AbstractDomainEvent) -> None:
    """Add a domain event to this aggregate.

    Args:
        event: The domain event to add.
    """
    self.__events.append(event)

remove_event(event)

Remove a domain event from this aggregate.

Parameters:

Name	Type	Description	Default
event	AbstractDomainEvent	The domain event to remove.	required

Source code in core/spakky-domain/src/spakky/domain/models/aggregate_root.py

def remove_event(self, event: AbstractDomainEvent) -> None:
    """Remove a domain event from this aggregate.

    Args:
        event: The domain event to remove.
    """
    self.__events.remove(event)

clear_events()

Clear all domain events from this aggregate.

Source code in core/spakky-domain/src/spakky/domain/models/aggregate_root.py

def clear_events(self) -> None:
    """Clear all domain events from this aggregate."""
    self.__events.clear()

options: show_root_heading: false

spakky.domain.models.entity

Entity model for domain-driven design.

This module provides AbstractEntity base class for DDD entities with identity, validation, and immutability enforcement.

CannotMonkeyPatchEntityError

Bases: AbstractSpakkyDomainError

Raised when attempting to add attributes not defined in the entity schema.

AbstractEntity

Bases: AbstractDomainModel, IEquatable, ABC

Base class for DDD entities with identity and validation.

Entities are objects with unique identity that maintain consistency through validation and prevent unauthorized modifications.

uid instance-attribute

Unique identifier for this entity.

version = field(default_factory=uuid7) class-attribute instance-attribute

Version identifier for optimistic locking.

created_at = field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute

Timestamp when entity was created.

updated_at = field(default_factory=(lambda: datetime.now(UTC))) class-attribute instance-attribute

Timestamp when entity was last updated.

next_id() abstractmethod classmethod

Generate next unique identifier for this entity type.

Returns:

Type	Description
EquatableT	New unique identifier.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

@classmethod
@abstractmethod
def next_id(cls) -> EquatableT:
    """Generate next unique identifier for this entity type.

    Returns:
        New unique identifier.
    """
    ...

validate() abstractmethod

Validate entity state.

Raises:

Type	Description
AbstractDomainValidationError	If validation fails.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

@abstractmethod
def validate(self) -> None:
    """Validate entity state.

    Raises:
        AbstractDomainValidationError: If validation fails.
    """
    ...

__eq__(other)

Compare entities by identity.

Parameters:

Name	Type	Description	Default
other	object	Object to compare with.	required

Returns:

Type	Description
bool	True if same entity type and uid.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

@override
def __eq__(self, other: object) -> bool:
    """Compare entities by identity.

    Args:
        other: Object to compare with.

    Returns:
        True if same entity type and uid.
    """
    if not isinstance(other, type(self)):
        return False
    return self.uid == other.uid

__hash__()

Compute hash based on entity identity.

Returns:

Type	Description
int	Hash of uid.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

@override
def __hash__(self) -> int:
    """Compute hash based on entity identity.

    Returns:
        Hash of uid.
    """
    return hash(self.uid)

__post_init__()

Validate entity after initialization.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

@override
def __post_init__(self) -> None:
    """Validate entity after initialization."""
    self.validate()
    self.__initialized = True

__setattr__(__name, __value)

Set attribute with validation and rollback on failure.

Automatically updates updated_at and version when any attribute (except metadata fields) is modified after initialization.

Parameters:

Name	Type	Description	Default
__name	str	Attribute name.	required
__value	Any	New value.	required

Raises:

Type	Description
CannotMonkeyPatchEntityError	If attribute not in schema.

Source code in core/spakky-domain/src/spakky/domain/models/entity.py

def __setattr__(self, __name: str, __value: Any) -> None:
    """Set attribute with validation and rollback on failure.

    Automatically updates `updated_at` and `version` when any attribute
    (except metadata fields) is modified after initialization.

    Args:
        __name: Attribute name.
        __value: New value.

    Raises:
        CannotMonkeyPatchEntityError: If attribute not in schema.
    """
    if __name not in self.__dataclass_fields__:
        raise CannotMonkeyPatchEntityError
    __old: Any | None = getattr(
        self, __name, None
    )  # 프레임워크 내부: 도메인 모델 변경 추적
    super().__setattr__(__name, __value)
    if self.__initialized:
        try:
            self.validate()
            # Auto-update metadata fields when business attributes change
            if __name not in self._AUTO_UPDATE_EXCLUDE_FIELDS:
                super().__setattr__("updated_at", datetime.now(UTC))
                super().__setattr__("version", uuid7())
        except:
            super().__setattr__(__name, __old)
            raise

options: show_root_heading: false

spakky.domain.models.value_object

Value object model for domain-driven design.

This module provides AbstractValueObject for representing immutable domain concepts compared by their attributes rather than identity.

UnhashableFieldTypeError(field_name)

Bases: AbstractSpakkyDomainError

Raised when a value object field type is not hashable.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

def __init__(self, field_name: str) -> None:
    self.field_name = field_name
    super().__init__(self.message)

AbstractValueObject

Bases: AbstractDomainModel, IEquatable, ICloneable, IDataclass, ABC

Base class for immutable value objects.

Value objects represent domain concepts without identity, compared by their attributes. All fields must be hashable.

clone()

Create copy of this value object.

Returns:

Type	Description
Self	Cloned value object.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@override
def clone(self) -> Self:
    """Create copy of this value object.

    Returns:
        Cloned value object.
    """
    return self

validate() abstractmethod

Validate value object state.

Raises:

Type	Description
AbstractDomainValidationError	If validation fails.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@abstractmethod
def validate(self) -> None:
    """Validate value object state.

    Raises:
        AbstractDomainValidationError: If validation fails.
    """
    ...

__eq__(__value)

Compare value objects by attributes.

Parameters:

Name	Type	Description	Default
__value	object	Object to compare with.	required

Returns:

Type	Description
bool	True if same type and all attributes equal.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@override
def __eq__(self, __value: object) -> bool:
    """Compare value objects by attributes.

    Args:
        __value: Object to compare with.

    Returns:
        True if same type and all attributes equal.
    """
    if not isinstance(__value, type(self)):
        return False
    return astuple(self) == astuple(__value)

__hash__()

Compute hash from all hashable attributes.

Returns:

Type	Description
int	Hash of tuple containing all attributes (order-preserving).

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@override
def __hash__(self) -> int:
    """Compute hash from all hashable attributes.

    Returns:
        Hash of tuple containing all attributes (order-preserving).
    """
    return hash(astuple(self))

__post_init__()

Validate value object after initialization.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@override
def __post_init__(self) -> None:
    """Validate value object after initialization."""
    self.validate()

__init_subclass__()

Verify all attributes are hashable.

Raises:

Type	Description
TypeError	If any attribute type is not hashable.

Source code in core/spakky-domain/src/spakky/domain/models/value_object.py

@override
def __init_subclass__(cls) -> None:
    """Verify all attributes are hashable.

    Raises:
        TypeError: If any attribute type is not hashable.
    """
    super().__init_subclass__()
    for name, type in cls.__annotations__.items():
        # typing 별칭/제네릭 타입은 __hash__ 속성이 없을 수 있어 안전 조회
        if getattr(type, "__hash__", None) is None:  # hashability protocol check
            raise UnhashableFieldTypeError(name)

options: show_root_heading: false

spakky.domain.models.event

Domain event models for event-driven architecture.

This module provides base classes for domain and integration events in event-driven systems.

AbstractEvent

Bases: AbstractDomainModel, IEquatable, IComparable, ICloneable, ABC

Base class for domain events.

event_id = field(default_factory=uuid4) class-attribute instance-attribute

Unique identifier for this event.

timestamp = field(default_factory=(lambda: datetime.now(timezone.utc))) class-attribute instance-attribute

When the event occurred.

event_name property

Get event type name.

Returns:

Type	Description
str	Class name of the event.

clone()

Create copy of this event.

Returns:

Type	Description
Self	Cloned event instance.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

def clone(self) -> Self:
    """Create copy of this event.

    Returns:
        Cloned event instance.
    """
    return self

__eq__(other)

Compare events by id and timestamp.

Parameters:

Name	Type	Description	Default
other	object	Object to compare with.	required

Returns:

Type	Description
bool	True if same event type, id, and timestamp.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __eq__(self, other: object) -> bool:
    """Compare events by id and timestamp.

    Args:
        other: Object to compare with.

    Returns:
        True if same event type, id, and timestamp.
    """
    if not isinstance(other, type(self)):
        return False
    return self.event_id == other.event_id and self.timestamp == other.timestamp

__hash__()

Compute hash from event id and timestamp.

Returns:

Type	Description
int	Hash of tuple containing event id and timestamp.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __hash__(self) -> int:
    """Compute hash from event id and timestamp.

    Returns:
        Hash of tuple containing event id and timestamp.
    """
    return hash((self.event_id, self.timestamp))

__lt__(__value)

Compare events by timestamp (less than).

Parameters:

Name	Type	Description	Default
__value	Self	Event to compare with.	required

Returns:

Type	Description
bool	True if this event occurred before the other.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __lt__(self, __value: Self) -> bool:
    """Compare events by timestamp (less than).

    Args:
        __value: Event to compare with.

    Returns:
        True if this event occurred before the other.
    """
    return self.timestamp < __value.timestamp

__le__(__value)

Compare events by timestamp (less than or equal).

Parameters:

Name	Type	Description	Default
__value	Self	Event to compare with.	required

Returns:

Type	Description
bool	True if this event occurred before or at same time as the other.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __le__(self, __value: Self) -> bool:
    """Compare events by timestamp (less than or equal).

    Args:
        __value: Event to compare with.

    Returns:
        True if this event occurred before or at same time as the other.
    """
    return self.timestamp <= __value.timestamp

__gt__(__value)

Compare events by timestamp (greater than).

Parameters:

Name	Type	Description	Default
__value	Self	Event to compare with.	required

Returns:

Type	Description
bool	True if this event occurred after the other.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __gt__(self, __value: Self) -> bool:
    """Compare events by timestamp (greater than).

    Args:
        __value: Event to compare with.

    Returns:
        True if this event occurred after the other.
    """
    return self.timestamp > __value.timestamp

__ge__(__value)

Compare events by timestamp (greater than or equal).

Parameters:

Name	Type	Description	Default
__value	Self	Event to compare with.	required

Returns:

Type	Description
bool	True if this event occurred after or at same time as the other.

Source code in core/spakky-domain/src/spakky/domain/models/event.py

@override
def __ge__(self, __value: Self) -> bool:
    """Compare events by timestamp (greater than or equal).

    Args:
        __value: Event to compare with.

    Returns:
        True if this event occurred after or at same time as the other.
    """
    return self.timestamp >= __value.timestamp

AbstractDomainEvent

Bases: AbstractEvent, ABC

Base class for domain events.

Domain events represent state changes within the domain that other parts of the system may be interested in.

AbstractIntegrationEvent

Bases: AbstractEvent, ABC

Base class for integration events.

Integration events are published across bounded contexts or services to communicate significant domain changes.

options: show_root_heading: false

애플리케이션 (CQRS)

spakky.domain.application.command

Command pattern abstractions for CQRS.

This module provides base classes and protocols for implementing command use cases in CQRS architecture.

AbstractCommand

Bases: ABC

Base class for command DTOs.

Commands represent intent to change system state.

ICommandUseCase

Bases: ABC

Interface for synchronous command use cases.

run(command) abstractmethod

Execute command and return result.

Parameters:

Name	Type	Description	Default
command	CommandT_contra	The command to execute.	required

Returns:

Type	Description
ResultT_co	Result of command execution.

Source code in core/spakky-domain/src/spakky/domain/application/command.py

@abstractmethod
def run(self, command: CommandT_contra) -> ResultT_co:
    """Execute command and return result.

    Args:
        command: The command to execute.

    Returns:
        Result of command execution.
    """
    ...

IAsyncCommandUseCase

Bases: ABC

Interface for asynchronous command use cases.

run(command) abstractmethod

Execute command asynchronously and return result.

The declaration uses Awaitable[ResultT_co] instead of async def so that the covariant ResultT_co remains sound under pyrefly's variance analysis. Concrete implementations may still use async def run because Coroutine is a subtype of Awaitable.

Parameters:

Name	Type	Description	Default
command	CommandT_contra	The command to execute.	required

Returns:

Type	Description
Awaitable[ResultT_co]	Awaitable resolving to the result of command execution.

Source code in core/spakky-domain/src/spakky/domain/application/command.py

@abstractmethod
def run(self, command: CommandT_contra) -> Awaitable[ResultT_co]:
    """Execute command asynchronously and return result.

    The declaration uses ``Awaitable[ResultT_co]`` instead of
    ``async def`` so that the covariant ``ResultT_co`` remains sound
    under pyrefly's variance analysis. Concrete implementations may
    still use ``async def run`` because ``Coroutine`` is a subtype of
    ``Awaitable``.

    Args:
        command: The command to execute.

    Returns:
        Awaitable resolving to the result of command execution.
    """
    ...

options: show_root_heading: false

spakky.domain.application.query

Query pattern abstractions for CQRS.

This module provides base classes and protocols for implementing query use cases in CQRS architecture.

AbstractQuery

Bases: ABC

Base class for query DTOs.

Queries represent intent to read system state without modification.

IQueryUseCase

Bases: ABC

Interface for synchronous query use cases.

run(query) abstractmethod

Execute query and return result.

Parameters:

Name	Type	Description	Default
query	QueryT_contra	The query to execute.	required

Returns:

Type	Description
ResultT_co	Query result.

Source code in core/spakky-domain/src/spakky/domain/application/query.py

@abstractmethod
def run(self, query: QueryT_contra) -> ResultT_co:
    """Execute query and return result.

    Args:
        query: The query to execute.

    Returns:
        Query result.
    """
    ...

IAsyncQueryUseCase

Bases: ABC

Interface for asynchronous query use cases.

run(query) abstractmethod

Execute query asynchronously and return result.

The declaration uses Awaitable[ResultT_co] instead of async def so that the covariant ResultT_co remains sound under pyrefly's variance analysis. Concrete implementations may still use async def run because Coroutine is a subtype of Awaitable.

Parameters:

Name	Type	Description	Default
query	QueryT_contra	The query to execute.	required

Returns:

Type	Description
Awaitable[ResultT_co]	Awaitable resolving to the query result.

Source code in core/spakky-domain/src/spakky/domain/application/query.py

@abstractmethod
def run(self, query: QueryT_contra) -> Awaitable[ResultT_co]:
    """Execute query asynchronously and return result.

    The declaration uses ``Awaitable[ResultT_co]`` instead of
    ``async def`` so that the covariant ``ResultT_co`` remains sound
    under pyrefly's variance analysis. Concrete implementations may
    still use ``async def run`` because ``Coroutine`` is a subtype of
    ``Awaitable``.

    Args:
        query: The query to execute.

    Returns:
        Awaitable resolving to the query result.
    """
    ...

options: show_root_heading: false

에러

spakky.domain.error

Error types for domain-level exceptions.

This module defines base error classes for domain logic and validation failures.

AbstractSpakkyDomainError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all domain-related errors.

AbstractDomainValidationError

Bases: AbstractSpakkyDomainError, ABC

Base class for domain validation errors.

options: show_root_heading: false