This guide demonstrates how to efficiently retrieve a large batch of drives using the MileIQ External API, including handling rate limits. We’ll use Python to showcase the process, but the principles can be applied to any programming language.

​
Prerequisites

Before you begin, make sure you have:

  1. A MileIQ API key
  2. Python 3.9 or later installed
  3. The requests library installed (pip install requests)
  4. The pytz library installed (pip install pytz)

​
Pagination Explanation

It’s important to note that the MileIQ External API doesn’t use traditional offset/limit pagination. Instead, it uses a timestamp-based pagination method. To paginate through the results, you need to use the modified_before and modified_after parameters in conjunction with the has_more field in the response body.

Here’s how it works:

  1. In your initial request, you set modified_after to the start of your desired date range and modified_before to the end of the range.
  2. The API returns a batch of results and a has_more boolean indicating if there are more results to fetch.
  3. If has_more is true, you make another request, but this time you set modified_before to the modified timestamp of the last drive in the previous batch.
  4. You repeat this process until has_more is false or you’ve retrieved all the drives you need.

This pagination method allows for efficient retrieval of large datasets while maintaining consistency even if new drives are added or modified during the pagination process.

​
Setting Up

First, let’s import the necessary libraries and set up our API configuration:

import requests
from datetime import datetime, timedelta
import pytz
import time

def make_api_request(url, params, headers):
    response = requests.get(url, params=params, headers=headers)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        print(f"Rate limit exceeded. Waiting for {retry_after} seconds.")
        time.sleep(retry_after)
        return make_api_request(url, params, headers)  # Retry the request

    # Print rate limiting information
    print(f"Rate Limit: {response.headers.get('X-RateLimit-Limit')}")
    print(f"Remaining: {response.headers.get('X-RateLimit-Remaining')}")
    print(f"Reset: {response.headers.get('X-RateLimit-Reset')}")

    return response

API_BASE_URL = "https://api.mileiq.com/v1/drives"
API_KEY = "your_api_key_here"

# Set the date range for which you want to retrieve drives
end_date = datetime.now(pytz.utc)
start_date = end_date - timedelta(days=30)  # Retrieve drives from the last 30 days

# Initialize variables
all_drives = []
has_more = True
modified_before = end_date.isoformat()

while has_more:
    # Prepare the request parameters
    params = {
        "modified_after": start_date.isoformat(),
        "modified_before": modified_before,
        "limit": 1000  # Maximum allowed limit
    }

    # Make the API request
    headers = {"Authorization": f"Bearer {API_KEY}"}
    response = make_api_request(API_BASE_URL, params, headers)

    if response.status_code == 200:
        data = response.json()
        drives = data["results"]
        has_more = data["has_more"]

        if drives:
            all_drives.extend(drives)
            # Update modified_before for the next iteration
            modified_before = drives[-1]["modified"]
        else:
            # No more drives in this time range
            has_more = False
    else:
        print(f"Error: {response.status_code} - {response.text}")
        break

    print(f"Retrieved {len(drives)} drives. Total drives: {len(all_drives)}")

    # Check if we're close to the rate limit
    remaining = int(response.headers.get('X-RateLimit-Remaining', 0))
    if remaining < 5:  # Arbitrary threshold, adjust as needed
        reset_time = int(response.headers.get('X-RateLimit-Reset', 0))
        wait_time = max(reset_time - int(time.time()), 0)
        print(f"Close to rate limit. Waiting for {wait_time} seconds.")
        time.sleep(wait_time)

print(f"Finished retrieving drives. Total drives retrieved: {len(all_drives)}")

## Process the retrieved drives

for drive in all_drives: # Here you can process each drive as needed
    print(f"Drive ID: {drive['id']}, Modified: {drive['modified']}")

## Example of how to use the retrieved data

if all_drives:
    print("\nExample of the first drive retrieved:")
    first_drive = all_drives[0]
    print(f"Driver: {first_drive['driver']['name']}")
    print(f"Distance: {first_drive['details']['distance']['amount']} {first_drive['details']['distance']['units']}")
    print(f"Value: {first_drive['details']['value']['amount']/100} {first_drive['details']['value']['currency']}")
    print(f"Status: {first_drive['status']}")
ChangelogGet Drives