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.

Example 
version = AggregateVersion(1)
assert version.value == 1
new_version = version.increment()
assert new_version.value == 2
Source code in src/forging_blocks/domain/aggregate_root/aggregate_version.py 
 6
 7
 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
35
class AggregateVersion(ValueObject[int]):
    """Immutable value object representing the version of an aggregate root.

    Used for optimistic concurrency control to detect conflicting updates.

    Example:
        ```python
        version = AggregateVersion(1)
        assert version.value == 1
        new_version = version.increment()
        assert new_version.value == 2
        ```
    """

    __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)

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 
33
34
35
def increment(self) -> AggregateVersion:
    """Return a new AggregateVersion incremented by one."""
    return AggregateVersion(self._value + 1)