Skip to content

sequence

django_spire.contrib.sync.django.sequence

SyncSequenceAllocator

Source code in django_spire/contrib/sync/django/sequence.py 
12
13
14
15
16
17
18
def __init__(self, counter_name: str = _DEFAULT_COUNTER_NAME, using: str | None = None) -> None:
    if not counter_name:
        message = 'counter_name must be a non-empty string'
        raise InvalidParameterError(message)

    self._counter_name = counter_name
    self._using = using

allocate

Source code in django_spire/contrib/sync/django/sequence.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
def allocate(self, count: int = 1) -> SequenceRange:
    if count < 1:
        message = f'count must be >= 1, got {count}'
        raise InvalidParameterError(message)

    counter = self._get_or_create_locked()

    value_first = counter.value + 1
    value_last = counter.value + count
    counter.value = value_last
    counter.save(update_fields=['value', 'updated_at'])

    return SequenceRange(value_first=value_first, value_last=value_last)

current

Source code in django_spire/contrib/sync/django/sequence.py 
48
49
50
51
52
53
54
55
56
57
58
def current(self) -> int:
    counter = (
        SyncSequenceCounter.objects
        .filter(name=self._counter_name)
        .first()
    )

    if counter is None:
        return 0

    return counter.value

reconcile_to

Source code in django_spire/contrib/sync/django/sequence.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def reconcile_to(self, value: int) -> None:
    if value < 1:
        return

    SyncSequenceCounter.objects.get_or_create(
        name=self._counter_name,
        defaults={'value': 0},
    )

    SyncSequenceCounter.objects.filter(
        name=self._counter_name,
        value__lt=value,
    ).update(value=value)