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

import typing
from abc import ABCMeta, abstractmethod

from ascetic_ddd.seedwork.domain.aggregate.interfaces import IVersionedAggregate

__all__ = (
    "VersionedAggregate",
    "IVersionedAggregateExporter",
    "IVersionedAggregateReconstitutor",
    "VersionedAggregateExporter",
    "VersionedAggregateReconstitutor",
)



[docs]
class VersionedAggregate(IVersionedAggregate, metaclass=ABCMeta):
    __version: int

    def __init__(self, version: int = 0, **kwargs) -> None:
        self.__version = version
        super().__init__(**kwargs)

    @property
    def version(self) -> int:
        return self.__version

    @version.setter
    def version(self, value: int) -> None:
        self.__version = value


[docs]
    def next_version(self) -> int:
        self.__version += 1
        return self.__version



[docs]
    def export(self, exporter: "IVersionedAggregateExporter") -> None:
        exporter.set_version(self.version)


    def _import(self, provider: "IVersionedAggregateReconstitutor") -> None:
        self.version = provider.version()

    @classmethod
    def _make_empty(cls) -> typing.Self:
        agg = cls.__new__(cls)
        super(cls, agg).__init__()
        return agg


[docs]
    @classmethod
    def reconstitute(cls, reconstitutor: "IVersionedAggregateReconstitutor") -> typing.Self:
        """
        For EventSourcedAggregate this method could be used to restore a snapshot.
        """
        agg: typing.Self = cls._make_empty()
        agg._import(reconstitutor)
        return agg





[docs]
class IVersionedAggregateExporter(metaclass=ABCMeta):

[docs]
    @abstractmethod
    def set_version(self, value: int) -> None:
        raise NotImplementedError





[docs]
class IVersionedAggregateReconstitutor(typing.Protocol, metaclass=ABCMeta):

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





[docs]
class VersionedAggregateExporter(IVersionedAggregateExporter):
    def __init__(self):
        self.data = dict()


[docs]
    def set_version(self, value: int) -> None:
        self.data['version'] = value





[docs]
class VersionedAggregateReconstitutor(IVersionedAggregateReconstitutor):
    def __init__(self, version: int):
        self._data = locals()


[docs]
    def version(self) -> int:
        return self._data['version']