enforce_rl

Signed-off-by: Finbarrs Oketunji <f@finbarrs.eu>

Author: 0xnu

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 1.0.2 - 2023-09-25
+* Added:
+  + ratelimit.py
+
+* Modified:
+  + __init__.py
+  + motdata.py
+  + test.py
+  + setup.py
+  + VERSION
+
 ## 1.0.1 - 2023-09-21
 * Updated: test.py, README.md, and LONG_DESCRIPTION.rst
 

VERSION

@@ -1 +1 @@
-1.0.1
+1.0.2

motapi/__init__.py

@@ -14,4 +14,4 @@
     "MOTHistoryAPI"
 ]
 
-__version__ = "1.0.1"
+__version__ = "1.0.2"

motapi/motdata.py

@@ -1,8 +1,9 @@
 """MOT Data API Client for interacting with the MOT history API."""
 
 from typing import Dict, Optional
+from .ratelimit import limits, sleep_and_retry, RateLimitExceeded
 import requests
-
+import time
 
 class MOTHistoryAPI:
     """Base class for interacting with the MOT history API."""
@@ -11,6 +12,11 @@ class MOTHistoryAPI:
     TOKEN_URL = "https://login.microsoftonline.com/a455b827-244f-4c97-b5b4-ce5d13b4d00c/oauth2/v2.0/token"
     SCOPE_URL = "https://tapi.dvsa.gov.uk/.default"
 
+    # Rate Limiting
+    QUOTA_LIMIT = 500000  # Maximum number of requests per day
+    BURST_LIMIT = 10  # Maximum number of requests in a short burst
+    RPS_LIMIT = 15  # Average number of requests per second
+
     def __init__(self, client_id: str, client_secret: str, api_key: str):
         """
         Initialise the MOT History API client.
@@ -24,6 +30,8 @@ def __init__(self, client_id: str, client_secret: str, api_key: str):
         self.client_secret = client_secret
         self.api_key = api_key
         self.access_token = self._get_access_token()
+        self.request_count = 0
+        self.last_request_time = time.time()
 
     def _get_access_token(self) -> str:
         """
@@ -57,9 +65,12 @@ def _get_headers(self) -> Dict[str, str]:
             "X-API-Key": self.api_key,
         }
 
+    @sleep_and_retry
+    @limits(calls=BURST_LIMIT, period=1)
+    @limits(calls=QUOTA_LIMIT, period=86400)
     def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
         """
-        Make a GET request to the API.
+        Make a GET request to the API with rate limiting.
 
         Args:
             endpoint (str): The API endpoint.
@@ -70,12 +81,23 @@ def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
 
         Raises:
             requests.exceptions.HTTPError: If the API request fails.
+            RateLimitExceeded: If the rate limit is exceeded.
         """
+        # RPS Limiting
+        current_time = time.time()
+        time_since_last_request = current_time - self.last_request_time
+        if time_since_last_request < 1 / self.RPS_LIMIT:
+            sleep_time = (1 / self.RPS_LIMIT) - time_since_last_request
+            raise RateLimitExceeded("RPS limit exceeded", sleep_time)
+
         url = f"{self.BASE_URL}/{endpoint}"
         response = requests.get(url, headers=self._get_headers(), params=params)
         response.raise_for_status()
-        return response.json()
 
+        self.request_count += 1
+        self.last_request_time = time.time()
+
+        return response.json()
 
 class VehicleData(MOTHistoryAPI):
     """Class for retrieving vehicle data."""
@@ -104,7 +126,6 @@ def get_by_vin(self, vin: str) -> Dict:
         """
         return self._make_request(f"vin/{vin}")
 
-
 class BulkData(MOTHistoryAPI):
     """Class for retrieving bulk MOT history data."""
 
@@ -117,7 +138,6 @@ def get_bulk_download(self) -> Dict:
         """
         return self._make_request("bulk-download")
 
-
 class CredentialsManager(MOTHistoryAPI):
     """Class for managing API credentials."""
 
@@ -143,7 +163,6 @@ def renew_client_secret(self, email: str) -> str:
         response.raise_for_status()
         return response.json()["clientSecret"]
 
-
 class MOTDataClient:
     """Main client class for interacting with the MOT history API."""
 

motapi/ratelimit.py

@@ -0,0 +1,87 @@
+"""The module provides rate-limiting decorators for functions and methods.
+
+It allows limiting the number of times a function can be called within a specified period.
+"""
+
+import time
+from functools import wraps
+
+
+class RateLimitExceeded(Exception):
+    """Exception raised when the rate limit is exceeded."""
+
+    def __init__(self, message: str, sleep_time: float):
+        """
+        Initialize the RateLimitExceeded exception.
+
+        Args:
+            message (str): The error message.
+            sleep_time (float): The time to sleep before retrying.
+        """
+        super().__init__(message)
+        self.sleep_time = sleep_time
+
+
+def sleep_and_retry(func):
+    """
+    A decorator that catches RateLimitExceeded exceptions and retries the function after sleeping.
+
+    Args:
+        func (callable): The function to be decorated.
+
+    Returns:
+        callable: The wrapped function that implements the sleep and retry behaviour.
+    """
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        while True:
+            try:
+                return func(*args, **kwargs)
+            except RateLimitExceeded as e:
+                time.sleep(e.sleep_time)
+    return wrapper
+
+
+def limits(calls=15, period=900, raise_on_limit=True):
+    """
+    A decorator factory that returns a decorator to rate limit function calls.
+
+    Args:
+        calls (int): The maximum number of calls allowed within the specified period.
+        period (int): The time period in seconds for which the calls are counted.
+        raise_on_limit (bool): If True, raises RateLimitExceeded when limit is reached.
+                               If False, blocks until the function can be called again.
+
+    Returns:
+        callable: A decorator that implements the rate limiting behaviour.
+    """
+    def decorator(func):
+        # Initialize the state for tracking function calls
+        func.__last_reset = time.monotonic()
+        func.__num_calls = 0
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            # Check if we need to reset the call count
+            now = time.monotonic()
+            time_since_reset = now - func.__last_reset
+            if time_since_reset > period:
+                func.__num_calls = 0
+                func.__last_reset = now
+
+            # Check if we've exceeded the rate limit
+            if func.__num_calls >= calls:
+                sleep_time = period - time_since_reset
+                if raise_on_limit:
+                    raise RateLimitExceeded("Rate limit exceeded", sleep_time)
+                else:
+                    time.sleep(sleep_time)
+                    return wrapper(*args, **kwargs)
+
+            # Call the function and increment the call count
+            func.__num_calls += 1
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator

setup.py

@@ -5,7 +5,7 @@
 from distutils.core import Extension
 
 NAME = "mot-history-api-py-sdk"
-VERSION = "1.0.1"
+VERSION = "1.0.2"
 REQUIRES = ["requests"]
 
 # read the contents of your README file

test.py

@@ -1,11 +1,12 @@
 """Test script for MOT Data API Client.
 
 It tests all the main functionalities of the MOT Data API Client,
-including handling various VIN lengths.
+including handling various VIN lengths and rate limiting scenarios.
 """
 
 import os
 import sys
+import time
 from pprint import pprint
 from typing import Optional
 from unittest.mock import Mock
@@ -14,8 +15,10 @@
 # but falls back to a mock if it is not present.
 try:
     from motapi import MOTDataClient
+    from motapi.ratelimit import RateLimitExceeded
 except ImportError:
     MOTDataClient = Mock()
+    RateLimitExceeded = Exception
 
 # Retrieve credentials from environment variables
 CLIENT_ID: Optional[str] = os.environ.get("MOT_CLIENT_ID")
@@ -108,6 +111,23 @@ def test_edge_case_vins():
         except Exception as e:
             print(f"Error retrieving data for VIN {vin}: {e}")
 
+def test_rate_limiting():
+    """Test rate limiting functionality."""
+    print("\nTesting rate limiting:")
+    start_time = time.time()
+    request_count = 0
+    try:
+        while time.time() - start_time < 5:  # Run for 5 seconds
+            client.get_vehicle_data("ML58FOU")
+            request_count += 1
+    except RateLimitExceeded as e:
+        print(f"Rate limit exceeded after {request_count} requests in {time.time() - start_time:.2f} seconds")
+        print(f"Rate limit exception details: {e}")
+    except Exception as e:
+        print(f"Unexpected error during rate limit testing: {e}")
+    else:
+        print(f"Made {request_count} requests in 5 seconds without hitting rate limit")
+
 if __name__ == "__main__":
     print("Starting MOT Data API Client tests...")
 
@@ -118,5 +138,6 @@ def test_edge_case_vins():
     test_renew_client_secret()
     test_invalid_vehicle_identifier()
     test_edge_case_vins()
+    test_rate_limiting()
 
     print("\nAll tests completed.")