Source code for ascetic_ddd.seedwork.domain.aggregate.interfaces

import typing
from abc import ABCMeta, abstractmethod

from ascetic_ddd.specification.domain.interfaces import IEqualOperand

__all__ = (
    "IVersionedAggregate",
    "IDomainEventAdder",
    "IDomainEventAccessor",
    "IEventiveEntity",
    "IDomainEventLoader",
    "IEventSourcedAggregate",
    "IHashable",
)



[docs]
class IVersionedAggregate(metaclass=ABCMeta):
    @property
    @abstractmethod
    def version(self) -> int:
        raise NotImplementedError

    @version.setter
    @abstractmethod
    def version(self, value: int) -> None:
        raise NotImplementedError


[docs]
    @abstractmethod
    def next_version(self) -> int:
        raise NotImplementedError




IDE = typing.TypeVar("IDE", covariant=True)



[docs]
class IDomainEventAdder(typing.Generic[IDE], metaclass=ABCMeta):
    @abstractmethod
    def _add_domain_event(self, event: IDE):
        raise NotImplementedError




[docs]
class IDomainEventAccessor(typing.Generic[IDE], metaclass=ABCMeta):
    @property
    @abstractmethod
    def pending_domain_events(self) -> typing.Iterable[IDE]:
        raise NotImplementedError

    @pending_domain_events.deleter
    @abstractmethod
    def pending_domain_events(self) -> None:
        raise NotImplementedError




[docs]
class IEventiveEntity(
    typing.Generic[IDE], IDomainEventAdder[IDE], IDomainEventAccessor[IDE], metaclass=ABCMeta
):
    pass



IPDE = typing.TypeVar("IPDE", covariant=True)



[docs]
class IDomainEventLoader(typing.Generic[IPDE], metaclass=ABCMeta):
    @abstractmethod
    def _load_from(self, past_events: typing.Iterable[IPDE]) -> None:
        raise NotImplementedError




[docs]
class IEventSourcedAggregate(
    typing.Generic[IPDE],
    IDomainEventLoader[IDomainEventLoader],
    IEventiveEntity[IPDE],
    IVersionedAggregate,
    metaclass=ABCMeta,
):
    @abstractmethod
    def _update(self, e: IPDE) -> None:
        raise NotImplementedError




[docs]
class IHashable(IEqualOperand, typing.Protocol):

    def __hash__(self) -> int:
        ...