feat(design, request-logger): logger rate limiter

DIRECTORY.md

@@ -514,6 +514,8 @@
   * Pubsub
     * Simple Events
       * [Test Simple Events](https://github.com/BrianLusina/PythonSnips/blob/master/design_patterns/pubsub/simple_events/test_simple_events.py)
+  * Request Logger
+    * [Test Request Logger](https://github.com/BrianLusina/PythonSnips/blob/master/design_patterns/request_logger/test_request_logger.py)
   * Structural
     * Proxy
       * Subject

design_patterns/request_logger/README.md

@@ -0,0 +1,67 @@
+# Logger Rate Limiter
+
+For the given stream of message requests and their timestamps as input, you must implement a logger rate limiter system 
+that decides whether the current message request is displayed. The decision depends on whether the same message has 
+already been displayed in the last S seconds. If yes, then the decision is FALSE, as this message is considered a 
+duplicate. Otherwise, the decision is TRUE.
+
+> Note: Several message requests, though received at different timestamps, may carry identical messages.
+
+## Constraints
+
+- 1 <= `request.length` <= 10^2
+- 0 <= `timestamp` <= 10^3
+- Timestamps are in ascending order.
+- Messages can be written in lowercase or uppercase English alphabets.
+
+## Examples
+
+![Example 1](./images/examples/request_logger_example_1.png)
+![Example 2](./images/examples/request_logger_example_2.png)
+![Example 3](./images/examples/request_logger_example_3.png)
+
+## Solution
+
+We need to know if a message already exists and keep track of its time limit. When thinking about such problems where 
+two associated values need to be checked, we can use a hash map.
+
+We can use all incoming messages as keys and their respective time limits as values. This will help us eliminate 
+duplicates and respect the time limit of S seconds as well.
+
+Here is how we’ll implement our algorithm using hash maps:
+
+1. Initialize a hash map. 
+2. When a request arrives, check if it’s a new request (the message is not among the keys stored in the hash map) or a 
+   repeated request (an entry for this message already exists in the hash map). If it’s a new request, accept it and add 
+   it to the hash map. 
+3. If it’s a repeated request, compare the difference between the timestamp of the incoming request and the timestamp of 
+   the previous request with the same message. If this difference is greater than the time limit, S, accept it and 
+   update the timestamp for that specific message in the hash map. Otherwise, reject it.
+
+![Solution 1](./images/solutions/request_logger_solution_1.png)
+![Solution 2](./images/solutions/request_logger_solution_2.png)
+![Solution 3](./images/solutions/request_logger_solution_3.png)
+![Solution 4](./images/solutions/request_logger_solution_4.png)
+![Solution 5](./images/solutions/request_logger_solution_5.png)
+
+### Solution Summary
+
+Let’s summarize our optimized algorithm:
+
+1. After initializing a hash map, whenever a request arrives, we check whether it’s a new request or a repeated request
+   after the assigned time limit
+
+2. If the request meets either of the conditions mentioned in the above step, we accept and update the entry associated
+   with that specific request in the hash map. Otherwise, reject the request and return the final decision.
+
+### Time Complexity
+
+The decision function checks whether a message has already been encountered, and if so, how long ago it was encountered.
+Thanks to the use of hash maps, both operations are completed in constant time—therefore, the time complexity of the
+decision function is O(1).
+
+### Space Complexity
+
+The space complexity of the algorithm is O(n), where n is the number of incoming requests that we store.
+
+

design_patterns/request_logger/__init__.py

@@ -0,0 +1,23 @@
+from typing import Dict
+
+
+class RequestLogger:
+    def __init__(self, time_limit: int):
+        self.time_limit = time_limit
+        # keeps track of the requests as they come in key value pairs, which allows for first lookups (O(1)).
+        # the key is the request, the value is the time.
+        self.request_map: Dict[str, int] = {}
+
+    # This function decides whether the message request should be accepted or rejected
+    def message_request_decision(self, timestamp: int, request: str) -> bool:
+        formatted_message = request.lower()
+        if formatted_message in self.request_map:
+            latest_time_for_message = self.request_map[formatted_message]
+            difference = timestamp - latest_time_for_message
+            if difference < self.time_limit:
+                return False
+            self.request_map[formatted_message] = timestamp
+            return True
+
+        self.request_map[formatted_message] = timestamp
+        return True

design_patterns/request_logger/test_request_logger.py

@@ -0,0 +1,43 @@
+import unittest
+from typing import List, Tuple
+from parameterized import parameterized
+from design_patterns.request_logger import RequestLogger
+
+REQUEST_LOGGER_TEST_CASES = [
+    (7, [(1, "Good morning", True), (6, "Good morning", False)]),
+    (7, [(4, "Hello world", True), (15, "Hello world", True)]),
+    (7, [(1, "Hello world", True), (2, "Good morning", True)]),
+    (
+        7,
+        [
+            (1, "good morning", True),
+            (5, "good morning", False),
+            (9, "i need coffee", True),
+            (10, "hello world", True),
+            (11, "good morning", True),
+            (15, "i need coffee", False),
+            (17, "hello world", True),
+            (25, "i need coffee", True),
+        ],
+    ),
+]
+
+
+class RequestLoggerTestCases(unittest.TestCase):
+    @parameterized.expand(REQUEST_LOGGER_TEST_CASES)
+    def test_request_logger(
+        self, time_limit: int, requests: List[Tuple[int, str, bool]]
+    ):
+        request_logger = RequestLogger(time_limit)
+        for input_request in requests:
+            timestamp, request, expected = input_request
+            actual = request_logger.message_request_decision(timestamp, request)
+            self.assertEqual(
+                expected,
+                actual,
+                f"Expected {expected}, but got {actual} for timestamp={timestamp}, request={request}",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()