def _update_token_count()

in google/cloud/sql/connector/rate_limiter.py [0:0]

• 6 lines of code
• 1 McCabe index (conditional complexity)

    def _update_token_count(self) -> None:
        """
        Calculates how much time has passed since the last leak and removes the
        appropriate amount of events from the queue.
        Leaking is done lazily, meaning that if there is a large time gap between
        leaks, the next set of calls might be a burst if burst_size > 1
        """
        now = self._loop.time()
        time_elapsed = now - self._last_token_update
        new_tokens = time_elapsed * self.rate
        self._tokens = min(new_tokens + self._tokens, self.max_capacity)
        self._last_token_update = now