Gecko [ www.jobworksschool.com ]

Linux jobworks 6.8.0-136-generic #136-Ubuntu SMP PREEMPT_DYNAMIC Wed Jul 1 21:53:05 UTC 2026 x86_64
Apache/2.4.58 (Ubuntu)
Server IP : 10.0.1.5 & Your IP : 216.73.217.52
Domains : Cant Read [ /etc/named.conf ]
User : www-data

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2025-08-05 17:14
__init__.py	121 B	-rw-r--r--	2025-08-05 17:14
adaptive.py	4.11 KB	-rw-r--r--	2025-08-05 17:14
base.py	797 B	-rw-r--r--	2025-08-05 17:14
bucket.py	3.9 KB	-rw-r--r--	2025-08-05 17:14
quota.py	1.89 KB	-rw-r--r--	2025-08-05 17:14
special.py	1.62 KB	-rw-r--r--	2025-08-05 17:14
standard.py	19.51 KB	-rw-r--r--	2025-08-05 17:14
throttling.py	1.74 KB	-rw-r--r--	2025-08-05 17:14

Rename

"""Retry quota implementation.

"""
import threading

class RetryQuota:
    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