Skip to content

Aggregate Version

aggregate_version

AggregateVersion value object for optimistic concurrency control.

AggregateVersion

Bases: ValueObject[int]

Immutable value object representing the version of an aggregate root.

Used for optimistic concurrency control to detect conflicting updates.

Source code in src/forging_blocks/domain/aggregate_root/aggregate_version.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class AggregateVersion(ValueObject[int]):
    """Immutable value object representing the version of an aggregate root.

    Used for optimistic concurrency control to detect conflicting updates.
    """

    __slots__ = ("_value",)

    def __init__(self, value: int) -> None:
        super().__init__()
        if value < 0:
            raise ValueError("Version cannot be negative")
        self._value = value

    @property
    def value(self) -> int:
        """Return the integer version value."""
        return self._value

    def increment(self) -> AggregateVersion:
        """Return a new AggregateVersion incremented by one."""
        return AggregateVersion(self._value + 1)

    @property
    def _equality_components(self) -> tuple[Hashable, ...]:
        """Components used for equality comparison."""
        return (self._value,)

value: int property

Return the integer version value.

increment() -> AggregateVersion

Return a new AggregateVersion incremented by one.

Source code in src/forging_blocks/domain/aggregate_root/aggregate_version.py 
27
28
29
def increment(self) -> AggregateVersion:
    """Return a new AggregateVersion incremented by one."""
    return AggregateVersion(self._value + 1)