-
-
Notifications
You must be signed in to change notification settings - Fork 59
/
limits.py
181 lines (131 loc) · 4.48 KB
/
limits.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
"""
"""
from __future__ import annotations
from functools import total_ordering
from typing import Dict, NamedTuple, Optional, Tuple, Type, Union, cast
def safe_string(value: Union[bytes, str, int]) -> str:
"""
converts a byte/str or int to a str
"""
if isinstance(value, bytes):
return value.decode()
return str(value)
class Granularity(NamedTuple):
seconds: int
name: str
TIME_TYPES = dict(
day=Granularity(60 * 60 * 24, "day"),
month=Granularity(60 * 60 * 24 * 30, "month"),
year=Granularity(60 * 60 * 24 * 30 * 12, "year"),
hour=Granularity(60 * 60, "hour"),
minute=Granularity(60, "minute"),
second=Granularity(1, "second"),
)
GRANULARITIES: Dict[str, Type[RateLimitItem]] = {}
class RateLimitItemMeta(type):
def __new__(
cls, name: str, parents: Tuple[type, ...], dct: Dict[str, Granularity]
) -> RateLimitItemMeta:
granularity = super().__new__(cls, name, parents, dct)
if "GRANULARITY" in dct:
GRANULARITIES[dct["GRANULARITY"][1]] = cast(
Type[RateLimitItem], granularity
)
return granularity
# pylint: disable=no-member
@total_ordering
class RateLimitItem(metaclass=RateLimitItemMeta):
"""
defines a Rate limited resource which contains the characteristic
namespace, amount and granularity multiples of the rate limiting window.
:param amount: the rate limit amount
:param multiples: multiple of the 'per' :attr:`GRANULARITY`
(e.g. 'n' per 'm' seconds)
:param namespace: category for the specific rate limit
"""
__slots__ = ["namespace", "amount", "multiples", "granularity"]
GRANULARITY: Granularity
"""
A tuple describing the granularity of this limit as
(number of seconds, name)
"""
def __init__(
self, amount: int, multiples: Optional[int] = 1, namespace: str = "LIMITER"
):
self.namespace = namespace
self.amount = int(amount)
self.multiples = int(multiples or 1)
@classmethod
def check_granularity_string(cls, granularity_string: str) -> bool:
"""
Checks if this instance matches a *granularity_string*
of type ``n per hour``, ``n per minute`` etc,
by comparing with :attr:`GRANULARITY`
"""
return granularity_string.lower() in cls.GRANULARITY.name
def get_expiry(self) -> int:
"""
:return: the duration the limit is enforced for in seconds.
"""
return self.GRANULARITY.seconds * self.multiples
def key_for(self, *identifiers: str) -> str:
"""
Constructs a key for the current limit and any additional
identifiers provided.
:param identifiers: a list of strings to append to the key
:return: a string key identifying this resource with
each identifier appended with a '/' delimiter.
"""
remainder = "/".join(
[safe_string(k) for k in identifiers]
+ [
safe_string(self.amount),
safe_string(self.multiples),
self.GRANULARITY.name,
]
)
return f"{self.namespace}/{remainder}"
def __eq__(self, other: object) -> bool:
if isinstance(other, RateLimitItem):
return self.amount == other.amount and self.GRANULARITY == other.GRANULARITY
return False
def __repr__(self) -> str:
return "%d per %d %s" % (self.amount, self.multiples, self.GRANULARITY.name)
def __lt__(self, other: RateLimitItem) -> bool:
return self.GRANULARITY.seconds < other.GRANULARITY.seconds
class RateLimitItemPerYear(RateLimitItem):
"""
per year rate limited resource.
"""
GRANULARITY = TIME_TYPES["year"]
"""A year"""
class RateLimitItemPerMonth(RateLimitItem):
"""
per month rate limited resource.
"""
GRANULARITY = TIME_TYPES["month"]
"""A month"""
class RateLimitItemPerDay(RateLimitItem):
"""
per day rate limited resource.
"""
GRANULARITY = TIME_TYPES["day"]
"""A day"""
class RateLimitItemPerHour(RateLimitItem):
"""
per hour rate limited resource.
"""
GRANULARITY = TIME_TYPES["hour"]
"""An hour"""
class RateLimitItemPerMinute(RateLimitItem):
"""
per minute rate limited resource.
"""
GRANULARITY = TIME_TYPES["minute"]
"""A minute"""
class RateLimitItemPerSecond(RateLimitItem):
"""
per second rate limited resource.
"""
GRANULARITY = TIME_TYPES["second"]
"""A second"""