How to Make Reliable HubSpot API Requests in Python (With Retry Logic)

Working with APIs can sometimes be tricky, especially when dealing with network timeouts, server errors, or rate limits.
In this post, we’ll walk through a Python function that makes requests to the HubSpot API with built-in retry logic to make it more reliable.

Why You Need Retry Logic for API Calls

When you send a request to HubSpot’s API, sometimes you’ll encounter:

Temporary server errors (500+)
Rate limits (HTTP 429)
Network connection issues
Timeouts due to slow response

{
    "status": "error",
    "message": "You have reached your secondly limit.",
    "errorType": "RATE_LIMIT",
    "correlationId": "fddeaec3-849f-408a-9c15-f62281e3a285",
    "policyName": "SECONDLY",
    "groupName": "publicapi:crm:search:oauth:2957456:39487040"
}

Instead of failing immediately, a good approach is to retry the request after a short delay, which is exactly what this function does.

The hubspot_request Function

Below is a reusable Python function that handles:

Automatic retries for transient errors
Exponential backoff (delay doubles after each retry)
Handling common HTTP status codes
Logging helpful error messages

import requests
import time
import json

# Dummy sample data
API_ACCESS_TOKEN = "your_api_access_token_here"
ENVIRONMENT = "prod"  # or "dev"

def log_status(status, message, extra=None):
    log = {"status": status, "message": message}
    if extra:
        log["extra"] = extra
    print(json.dumps(log))

def hubspot_request(url, method="GET", payload=None, params=None):
    """Send HTTP request to HubSpot API with retry logic."""
    
    retries = 4
    delay = 2 if ENVIRONMENT.lower() == "prod" else 10
    timeout = 30  # seconds

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_ACCESS_TOKEN}",
    }

    for attempt in range(retries):
        try:
            method = method.upper()
            request_args = {"headers": headers, "timeout": timeout}

            if params:
                request_args["params"] = params
            if payload:
                request_args["json"] = payload

            if method == "GET":
                response = requests.get(url, **request_args)
            elif method == "POST":
                response = requests.post(url, **request_args)
            elif method == "PATCH":
                response = requests.patch(url, **request_args)
            elif method == "PUT":
                response = requests.put(url, **request_args)
            elif method == "DELETE":
                response = requests.delete(url, **request_args)
            else:
                log_status("Error", f"Unsupported HTTP method: {method}")
                return None

            status = response.status_code

            if status in [200, 201]:
                return {
                    "statusCode": status,
                    "body": response.json(),
                    "message": "Request successful",
                    "method": method,
                    "url": url,
                }
            elif status == 204:
                return {}
            elif status >= 500:
                log_status("Error", f"Server error ({status}), retrying in {delay}s...", {
                    "url": url, "method": method, "payload": payload, "params": params
                })
                time.sleep(delay)
                delay *= 2
            elif status == 429:
                retry_after = int(response.headers.get("Retry-After", delay))
                log_status("Warning", f"Rate limit hit. Retrying in {retry_after}s...", {
                    "url": url, "method": method, "payload": payload, "params": params
                })
                time.sleep(retry_after)
                delay *= 2
            elif status in [401, 403, 404]:
                messages = {
                    401: "Unauthorized: Check your API key or token.",
                    403: "Forbidden: No permission to access this resource.",
                    404: "Not Found: The requested resource does not exist.",
                }
                log_status("Error", f"{messages[status]} ({status})", {
                    "url": url, "method": method, "payload": payload, "params": params
                })
                return None
            elif 400 <= status < 500:
                log_status("Error", f"Client Error ({status}): {response.text}", {
                    "url": url, "method": method, "payload": payload, "params": params
                })
                return None
            else:
                log_status("Error", f"Unhandled Status Code ({status})", {
                    "url": url, "method": method, "payload": payload, "params": params
                })
                return None

        except requests.exceptions.RequestException as e:
            log_status("Error", f"Request failed: {e}")
            time.sleep(delay)
            delay *= 2
        except requests.exceptions.ReadTimeout as e:
            log_status("Error", f"Request timed out: {e}")
            time.sleep(delay)
            delay *= 2
        except requests.exceptions.ConnectionError as e:
            log_status("Error", f"Connection error: {e}")
            time.sleep(delay)
            delay *= 2

    log_status("Error", f"Request failed after {retries} attempts. Last URL: {url}")
    return None

Key Features of This Function

Secure API Key Handling
Uses API_ACCESS_TOKEN to set your HubSpot token.
Supports Multiple HTTP Methods
Works with GET, POST, PATCH, PUT, and DELETE.
Automatic Retry on Failures
- Retries when server returns 500+ errors
- Waits before retrying to avoid spamming the server
- Doubles wait time after each failure
Rate Limit Handling (429)
Reads the Retry-After header from HubSpot and waits before trying again.
Structured Response
Returns a Python dictionary with:
- statusCode
- body (JSON data from HubSpot)
- message
- method and url

Example Usage

url = "https://api.hubapi.com/crm/v3/objects/contacts"
params = {"limit": 5}

response = hubspot_request(url, method="GET", params=params)

if response:
    print("Contacts retrieved:", response["body"])
else:
    print("Request failed.")

Conclusion

This hubspot_request function is a reliable and reusable way to call the HubSpot API without worrying too much about random failures or rate limits.
By adding retry logic, exponential backoff, and proper error handling, you can make your integrations more stable and easier to maintain.

📚 Further Learning

If you want to explore more, here are some valuable reads:

Sync HubSpot Company Records to S3 Using AWS Lambda and Step Functions – by Linuxbeast
How to Avoid AWS Lambda Timeout When Processing HubSpot Records – by Linuxbeast