pincer.utils.slidingwindow.SlidingWindow.allow() - Code Metrics - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

pincer.utils.slidingwindow.SlidingWindow.allow() A
last analyzed 2022-12-26 13:13 UTC

↳ Parent: pincer.utils.slidingwindow

Complexity

Conditions

Size

Total Lines	25
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	25
rs	9.8
c	0
b	0
f	0
cc	3
nop	1

# Copyright Pincer 2021-Present

# Full MIT License can be found in `LICENSE` at the project root.

from time import time


class SlidingWindow:

    def __init__(self, capacity: int, time_unit: float):
        self.capacity: int = capacity
        self.time_unit: float = time_unit

        self.__cur_time: float = time()
        self.__pre_count: int = capacity
        self.__cur_count: int = 0

    def allow(self) -> bool:
        # TODO: fix docs

        """

        Returns
        -------

        """
        # Reset rate limit:
        if (time() - self.__cur_time) > self.time_unit:
            self.__cur_time = time()
            self.__pre_count = self.__cur_count
            self.__cur_count = 0

        # Calculate the estimated count
        passed_time = time() - self.__cur_time
        time_cnt = (self.time_unit - passed_time) / self.time_unit
        est_cnt = self.__pre_count * time_cnt + self.__cur_count

        # Request has passed the capacity
        if est_cnt > self.capacity:
            return False

        self.__cur_count += 1
        return True


1			# Copyright Pincer 2021-Present
			0 ignored issues – show introduced 2021-09-22 03:25 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# Full MIT License can be found in `LICENSE` at the project root.
3
4			from time import time
5
6
7			class SlidingWindow:
			0 ignored issues – show introduced 2021-09-22 03:25 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
8			def __init__(self, capacity: int, time_unit: float):
9			self.capacity: int = capacity
10			self.time_unit: float = time_unit
11
12			self.__cur_time: float = time()
13			self.__pre_count: int = capacity
14			self.__cur_count: int = 0
15
16			def allow(self) -> bool:
17			# TODO: fix docs
			0 ignored issues – show Coding Style introduced 2021-11-20 22:37 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
18			"""
19
20			Returns
21			-------
22
23			"""
24			# Reset rate limit:
25			if (time() - self.__cur_time) > self.time_unit:
26			self.__cur_time = time()
27			self.__pre_count = self.__cur_count
28			self.__cur_count = 0
29
30			# Calculate the estimated count
31			passed_time = time() - self.__cur_time
32			time_cnt = (self.time_unit - passed_time) / self.time_unit
33			est_cnt = self.__pre_count * time_cnt + self.__cur_count
34
35			# Request has passed the capacity
36			if est_cnt > self.capacity:
37			return False
38
39			self.__cur_count += 1
40			return True
41

Pincer-org / Pincer

pincer.utils.slidingwindow.SlidingWindow.allow() A last analyzed 2022-12-26 13:13 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

pincer.utils.slidingwindow.SlidingWindow.allow() A
last analyzed 2022-12-26 13:13 UTC