-
Notifications
You must be signed in to change notification settings - Fork 17
/
Copy pathdesign-hit-counter.py
63 lines (50 loc) · 1.97 KB
/
design-hit-counter.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
from collections import deque
class TimestampCounter:
def __init__(self, timestamp, count):
self.timestamp = timestamp
self.count = count
def __repr__(self):
return str(f"TimestampCounter({self.timestamp}, {self.count})")
def current_time(timestamp):
return timestamp # FIXME: should be time.time() in real app
class HitCounter:
COUNTER_TIME_WINDOW = 300 # 5 minutes
def __init__(self):
"""
Initialize your data structure here.
"""
self.queue = deque(maxlen=self.COUNTER_TIME_WINDOW)
self.count = 0
self.timestamp_counters = {}
def hit(self, timestamp: int) -> None:
"""
Record a hit.
@param timestamp - The current timestamp (in seconds granularity).
"""
if timestamp in self.timestamp_counters:
timestamp_counter = self.timestamp_counters[timestamp]
timestamp_counter.count += 1
self.count += 1
elif timestamp > current_time(timestamp) - self.COUNTER_TIME_WINDOW:
# If we haven't received an event from the past
timestamp_counter = TimestampCounter(timestamp, 1)
self.queue.append(timestamp_counter)
self.timestamp_counters[timestamp] = timestamp_counter
self.count += 1
def getHits(self, timestamp: int) -> int:
"""
Return the number of hits in the past 5 minutes.
@param timestamp - The current timestamp (in seconds granularity).
"""
while (
self.queue
and self.queue[0].timestamp <= current_time(timestamp) - self.COUNTER_TIME_WINDOW
):
timestamp_counter = self.queue.popleft()
del self.timestamp_counters[timestamp_counter.timestamp]
self.count -= timestamp_counter.count
return self.count
# Your HitCounter object will be instantiated and called as such:
# obj = HitCounter()
# obj.hit(timestamp)
# param_2 = obj.getHits(timestamp)