spakky-domain

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

Models

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[IEquatable])) module-attribute

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

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

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

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

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

AbstractAggregateRoot

Bases: AbstractEntity[EquatableT_co], Generic[EquatableT_co], 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, Generic[EquatableT_co], 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_co	New unique identifier.

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

@classmethod
@abstractmethod
def next_id(cls) -> EquatableT_co:
    """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

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

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

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.

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

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

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

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

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

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():
        if getattr(type, "__hash__", None) is None:
            raise TypeError(f"type of '{name}' is not hashable")

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

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

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

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

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

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

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

Application (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.

CommandT_contra = TypeVar('CommandT_contra', bound=AbstractCommand, contravariant=True) module-attribute

Contravariant type variable for command types.

ResultT_co = TypeVar('ResultT_co', bound=Any, covariant=True) module-attribute

Covariant type variable for result types.

AbstractCommand

Bases: ABC

Base class for command DTOs.

Commands represent intent to change system state.

ICommandUseCase

Bases: ABC, Generic[CommandT_contra, ResultT_co]

Protocol 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, Generic[CommandT_contra, ResultT_co]

Protocol for asynchronous command use cases.

run(command) abstractmethod async

Execute command asynchronously 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
async def run(self, command: CommandT_contra) -> ResultT_co:  # pyrefly: ignore
    """Execute command asynchronously and return result.

    Args:
        command: The command to execute.

    Returns:
        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.

QueryT_contra = TypeVar('QueryT_contra', bound=AbstractQuery, contravariant=True) module-attribute

Contravariant type variable for query types.

ResultT_co = TypeVar('ResultT_co', bound=Any, covariant=True) module-attribute

Covariant type variable for result types.

AbstractQuery

Bases: ABC

Base class for query DTOs.

Queries represent intent to read system state without modification.

IQueryUseCase

Bases: ABC, Generic[QueryT_contra, ResultT_co]

Protocol 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, Generic[QueryT_contra, ResultT_co]

Protocol for asynchronous query use cases.

run(query) abstractmethod async

Execute query asynchronously 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
async def run(self, query: QueryT_contra) -> ResultT_co:  # pyrefly: ignore
    """Execute query asynchronously and return result.

    Args:
        query: The query to execute.

    Returns:
        Query result.
    """
    ...

options: show_root_heading: false

Errors

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