1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / botocore / retries /
Filename : quota.py
back
"""Retry quota implementation.


"""
import threading


class RetryQuota(object):
    INITIAL_CAPACITY = 500

    def __init__(self, initial_capacity=INITIAL_CAPACITY, lock=None):
        self._max_capacity = initial_capacity
        self._available_capacity = initial_capacity
        if lock is None:
            lock = threading.Lock()
        self._lock = lock

    def acquire(self, capacity_amount):
        """Attempt to aquire a certain amount of capacity.

        If there's not sufficient amount of capacity available, ``False``
        is returned.  Otherwise, ``True`` is returned, which indicates that
        capacity was successfully allocated.

        """
        # The acquire() is only called when we encounter a retryable
        # response so we aren't worried about locking the entire method.
        with self._lock:
            if capacity_amount > self._available_capacity:
                return False
            self._available_capacity -= capacity_amount
            return True

    def release(self, capacity_amount):
        """Release capacity back to the retry quota.

        The capacity being released will be truncated if necessary
        to ensure the max capacity is never exceeded.

        """
        # Implementation note:  The release() method is called as part
        # of the "after-call" event, which means it gets invoked for
        # every API call.  In the common case where the request is
        # successful and we're at full capacity, we can avoid locking.
        # We can't exceed max capacity so there's no work we have to do.
        if self._max_capacity == self._available_capacity:
            return
        with self._lock:
            amount = min(
                self._max_capacity - self._available_capacity,
                capacity_amount
            )
            self._available_capacity += amount

    @property
    def available_capacity(self):
        return self._available_capacity