Or: How I Learned to Stop Worrying and Love the Retry-After Header
If you've been building systems that rely on external APIs for more than a week, you've already felt it. That cold sweat when production breaks at 3 AM because Microsoft decided to throttle your tenant. That sinking feeling when a "beta" API you've relied on for months returns a 500 with no explanation. That special rage when you realize you need to reverse-engineer an SDK because the feature you need isn't officially documented.
Welcome to the club. Membership is involuntary, and the only exit is retirement.
I've spent years integrating with external APIs—Microsoft Graph, Mailgun, AWS SES, and a dozen others I've tried to forget. What follows isn't theory from a conference talk. It's scar tissue converted into wisdom. These are the battles I've fought, the mistakes I've made, and the patterns I've learned to survive when your system's reliability depends on someone else's infrastructure.
Think of this as the Blade Runner of API integration guides — dark, gritty, and questioning what it means to be human when you're debugging OAuth flows at 4 AM.
1. Rate Limiting: The Art of Not Pissing Off Your Provider
Let's start with Microsoft Graph, because they've perfected the art of making rate limiting a mystery religion. It's like trying to understand the plot of Dune after three glasses of ouzo — you think you get it, but you really don't.
Microsoft doesn't give you a strict number. No "100 requests per minute per tenant" that you can work with. Instead, you get vague guidance about "throttling when you hit a tenant too hard" and a promise that larger tenants have higher tolerance. Spoiler: they don't. We learned this the hard way when our enterprise customer with 5,000 users throttled faster than our 50-user test tenant.
Here's what actually works:
Respect the Retry-After header religiously. When you get a 429 (Too Many Requests) or 503 (Service Unavailable), Microsoft tells you exactly when to try again. Parse that header, wait that long, then retry. Not before. Not "eh, let's try 10 seconds." The exact amount. This isn't The Karate Kid—you can't just wax on, wax off and hope it works.
import time
import requests
from typing import Optional
async def graph_request(url: str, headers: dict, max_retries: int = 3) -> requests.Response:
"""Make a Graph API request with proper retry handling."""
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=30)
# Microsoft is retriable on 429 (throttling) or 503 (service unavailable)
if response.status_code in [429, 503]:
retry_after = int(response.headers.get('Retry-After', 5))
print(f"Throttled. Waiting {retry_after} seconds...")
time.sleep(retry_after)
continue
return response
# If we exhausted retries, raise
response.raise_for_status()
return response
Handle 401 with token refresh. A 401 doesn't mean your request is bad — it means your token expired. Refresh it and try ONE more time. If you get another 401, then you've got real problems:
class GraphClient:
def __init__(self, tenant_id: str, client_id: str, client_secret: str):
self.tenant_id = tenant_id
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
self.token_expires_at = 0
def get_token(self) -> str:
"""Get a fresh access token."""
if time.time() < self.token_expires_at:
return self.access_token
token_url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
data = {
'grant_type': 'client_credentials',
'client_id': self.client_id,
'client_secret': self.client_secret,
'scope': 'https://graph.microsoft.com/.default'
}
response = requests.post(token_url, data=data)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data['access_token']
self.token_expires_at = time.time() + token_data['expires_in'] - 60 # Refresh 60s early
return self.access_token
def request(self, url: str, max_retries: int = 3) -> requests.Response:
"""Make a Graph API request with token refresh on 401."""
headers = {'Authorization': f'Bearer {self.get_token()}'}
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=30)
# Handle token expiration
if response.status_code == 401:
if attempt == 0:
# First 401: refresh token and retry
print("Token expired, refreshing...")
headers = {'Authorization': f'Bearer {self.get_token()}'}
continue
else:
# Second 401: real auth problem, fail
raise Exception(f"Authentication failed even after token refresh: {response.text}")
# Handle throttling
if response.status_code in [429, 503]:
retry_after = int(response.headers.get('Retry-After', 5))
print(f"Throttled. Waiting {retry_after} seconds...")
time.sleep(retry_after)
continue
return response
response.raise_for_status()
return response
Global locking for multi-process systems. When you're running multiple workers hitting the same tenant, one worker getting throttled means all workers need to back off. It's like when the whole village needs to know that the one-eyed monster is coming — you need a signal everyone can see.
We used Redis for a global throttle flag:
import redis
from datetime import timedelta
class ThrottleManager:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def is_throttled(self, tenant_id: str) -> tuple[bool, int]:
"""Check if tenant is throttled and for how long."""
key = f"throttle:tenant:{tenant_id}"
ttl = self.redis.ttl(key)
if ttl > 0:
return True, ttl
return False, 0
def set_throttle(self, tenant_id: str, seconds: int):
"""Mark tenant as throttled for specified seconds."""
key = f"throttle:tenant:{tenant_id}"
self.redis.setex(key, timedelta(seconds=seconds), value='throttled')
async def wait_if_throttled(self, tenant_id: str):
"""Wait if tenant is currently throttled."""
is_throttled, wait_time = self.is_throttled(tenant_id)
if is_throttled:
print(f"Tenant {tenant_id} throttled. Waiting {wait_time}s...")
time.sleep(wait_time)
# Usage in worker
throttle_mgr = ThrottleManager(redis.Redis(host='localhost'))
async def process_user(tenant_id: str, user_id: str):
# Check global throttle before making request
await throttle_mgr.wait_if_throttled(tenant_id)
try:
response = graph_client.request(f"https://graph.microsoft.com/v1.0/users/{user_id}")
return response.json()
except requests.HTTPError as e:
if e.response.status_code in [429, 503]:
retry_after = int(e.response.headers.get('Retry-After', 5))
throttle_mgr.set_throttle(tenant_id, retry_after)
raise
Real-time vs. background strategies. In user-facing scenarios, if Retry-After is under 3 seconds, we retry and show a loading spinner. Over 3 seconds? Show the user "Microsoft's servers are busy, try again in a moment" and move on. Like when you're waiting for the Death Star plans to download — if it's taking too long, abort the mission.
For background jobs, we ran experiments. Started with 10 workers, got throttled constantly. Dropped to 5, still throttled. Settled on 3 workers with 2-second delays between requests. Not fast, but reliable. Your mileage will vary — run your own experiments.
from concurrent.futures import ThreadPoolExecutor
import time
def background_sync_with_delay(users: list[str], workers: int = 3, delay_between: float = 2.0):
"""Process users with controlled concurrency and delays."""
def process_with_delay(user_id: str):
result = process_user(tenant_id, user_id)
time.sleep(delay_between) # Throttle ourselves
return result
with ThreadPoolExecutor(max_workers=workers) as executor:
results = list(executor.map(process_with_delay, users))
return results
The nuclear option: Multiple OAuth apps. When nothing else works, create multiple OAuth applications. Each has its own rate limit pool. Rotate through them round-robin style. It's like having multiple fake IDs to get into the same club — frowned upon, but effective:
from itertools import cycle
class MultiAppGraphClient:
def __init__(self, apps: list[dict]):
"""
apps: list of dicts with 'client_id', 'client_secret', 'tenant_id'
"""
self.clients = [GraphClient(**app) for app in apps]
self.client_pool = cycle(self.clients)
def get_next_client(self) -> GraphClient:
"""Get next client in round-robin fashion."""
return next(self.client_pool)
def request(self, url: str) -> requests.Response:
"""Make request using next available client."""
client = self.get_next_client()
return client.request(url)
# Usage
apps = [
{'tenant_id': 'xxx', 'client_id': 'app1', 'client_secret': 'secret1'},
{'tenant_id': 'xxx', 'client_id': 'app2', 'client_secret': 'secret2'},
{'tenant_id': 'xxx', 'client_id': 'app3', 'client_secret': 'secret3'}
]
multi_client = MultiAppGraphClient(apps)
response = multi_client.request('https://graph.microsoft.com/v1.0/users')
Is this elegant? No. Does it work? Absolutely. Sometimes you need more buckets, not better buckets. It's the MacGyver solution — use what you have to solve the problem.
2. Service Downtime: When "Highly Available" Means "Usually Available"
External APIs go down. Accept this truth like you accept that Highlander 2 was a mistake. Microsoft goes down. AWS goes down. Your scrappy startup API definitely goes down. Build for it.
Error classification is survival. You need to understand three types of errors:
- Transient errors (429, 500, 502, 503, 504): Retry these
- Permanent errors (400, 403, 404): Don't retry these
- No response at all: The worst kind
- 200 OK with error in body: The sneaky bastard - legacy or poorly written APIs that return HTTP 200 but include {"error": "something went wrong"} in the response body
That last two nearly killed us. Early on, we didn't set timeouts on API calls. Graph would hang, our code would wait, and users would stare at loading screens for minutes. Then the thread pool would exhaust, and the whole service would stop responding. It was like waiting for Sarah Connor to answer the phone — eventually, you need to give up and run. That last one is my personal favorite. Nothing says "we don't understand HTTP status codes" quite like returning 200 OK when everything is definitely not okay.
Always set aggressive timeouts:
import requests
from requests.exceptions import Timeout, RequestException
class TransientError(Exception):
"""Retriable error."""
pass
class PermanentError(Exception):
"""Non-retriable error."""
pass
def api_call_with_timeout(url: str, timeout: int = 30) -> dict:
"""Make API call with timeout."""
try:
response = requests.get(url, timeout=timeout)
# Check for transient errors
if response.status_code in [429, 500, 502, 503, 504]:
raise TransientError(f"Transient error: {response.status_code}")
# Check for permanent errors
if response.status_code in [400, 403, 404]:
raise PermanentError(f"Permanent error: {response.status_code} - {response.text}")
response.raise_for_status()
return response.json()
except Timeout:
# Treat timeout as transient error
raise TransientError(f"Request timeout after {timeout}s")
except RequestException as e:
# Network errors are usually transient
raise TransientError(f"Network error: {str(e)}")
30 seconds was our sweet spot. Short enough that users don't give up, long enough that slow APIs can respond. Test with your specific APIs — some need more, some need less.
Always check for error in successful response and apply defensive parsing:
def api_call_with_timeout(url: str, timeout: int = 30) -> dict:
"""Make API call with timeout."""
try:
response = requests.get(url, timeout=timeout)
# Check for transient errors
if response.status_code in [429, 500, 502, 503, 504]:
raise TransientError(f"Transient error: {response.status_code}")
# Check for permanent errors
if response.status_code in [400, 403, 404]:
raise PermanentError(f"Permanent error: {response.status_code} - {response.text}")
response.raise_for_status()
# Parse response
data = response.json()
# Check for the "200 OK but actually error" pattern
# Different APIs do this differently, so check all common patterns
if 'error' in data:
error_msg = data.get('error')
# Try to classify if it's transient or permanent
if isinstance(error_msg, dict):
error_code = error_msg.get('code', '')
error_message = error_msg.get('message', '')
else:
error_code = ''
error_message = str(error_msg)
# Some errors are transient even with 200 OK
if any(word in error_message.lower() for word in ['timeout', 'temporarily', 'try again']):
raise TransientError(f"API returned error in body: {error_message}")
else:
raise PermanentError(f"API returned error in body: {error_message}")
# Check other common error patterns
if data.get('success') is False or data.get('status') == 'error':
error_message = data.get('message', 'Unknown error')
raise PermanentError(f"API indicated failure: {error_message}")
return data
except Timeout:
# Treat timeout as transient error
raise TransientError(f"Request timeout after {timeout}s")
except RequestException as e:
# Network errors are usually transient
raise TransientError(f"Network error: {str(e)}")This defensive parsing saved us more times than I can count. Some APIs would return {"success": false, "error": "rate limit exceeded"} with HTTP 200. Others would return {"status": "error", "message": "try again later"}. You need to check for all of them.
Build a retry wrapper with exponential backoff:
import random
import time
from functools import wraps
def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
"""Decorator for retrying functions with exponential backoff."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except TransientError as e:
if attempt == max_retries - 1:
# Last attempt, give up
raise
# Exponential backoff with jitter
delay = min(base_delay * (2 ** attempt), 30)
jitter = random.uniform(0, 1)
sleep_time = delay + jitter
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {sleep_time:.2f}s...")
time.sleep(sleep_time)
except PermanentError:
# Don't retry permanent errors
raise
return wrapper
return decorator
# Usage
@retry_with_backoff(max_retries=3)
def get_user(user_id: str) -> dict:
return api_call_with_timeout(f"https://graph.microsoft.com/v1.0/users/{user_id}")
# Call it
try:
user = get_user("some-user-id")
except PermanentError as e:
print(f"Permanent failure: {e}")
except TransientError as e:
print(f"Failed after retries: {e}")
The jitter (random delay) prevents thundering herd when many requests retry simultaneously. Without it, you get the API equivalent of everyone trying to exit the burning building at the same time.
3. Pagination: Everyone Does It Differently, Apparently
Every API pagination is a unique snowflake of pain. It's like every vendor watched a different 80s movie and decided to base their pagination on it:
Microsoft Graph: Next link in response body (Back to the Future style — follow the breadcrumbs)
def paginate_graph(url: str, client: GraphClient) -> list[dict]:
"""Paginate through Graph API results."""
all_items = []
next_url = url
while next_url:
response = client.request(next_url).json()
all_items.extend(response.get('value', []))
next_url = response.get('@odata.nextLink')
return all_items
# Usage
users = paginate_graph('https://graph.microsoft.com/v1.0/users', graph_client)
AWS APIs: Pagination token in response body (Indiana Jones style — you need the token to proceed)
import boto3
def paginate_aws(client, operation: str, **kwargs) -> list[dict]:
"""Paginate through AWS API results."""
all_items = []
next_token = None
while True:
if next_token:
kwargs['NextToken'] = next_token
response = getattr(client, operation)(**kwargs)
all_items.extend(response.get('Items', []))
next_token = response.get('NextToken')
if not next_token:
break
return all_items
# Usage
s3 = boto3.client('s3')
objects = paginate_aws(s3, 'list_objects_v2', Bucket='my-bucket')
Link header pagination: Uses Link header with rel="next" (The Terminator style—follow if present, stop if not)
import requests
from typing import Optional
def parse_link_header(link_header: Optional[str]) -> Optional[str]:
"""Parse Link header to extract next URL."""
if not link_header:
return None
links = link_header.split(',')
for link in links:
parts = link.split(';')
if len(parts) == 2 and 'rel="next"' in parts[1]:
return parts[0].strip('<> ')
return None
def paginate_with_link_header(url: str) -> list[dict]:
"""Paginate using Link headers (GitHub style)."""
all_items = []
next_url = url
while next_url:
response = requests.get(next_url, timeout=30)
response.raise_for_status()
all_items.extend(response.json())
next_url = parse_link_header(response.headers.get('Link'))
return all_items
# Usage
repos = paginate_with_link_header('https://api.github.com/users/someuser/repos')
Cursor-based pagination: Uses an opaque cursor (The Matrix style — you don't understand it, you just follow it)
def paginate_with_cursor(base_url: str, initial_params: dict = None) -> list[dict]:
"""Paginate using cursor."""
all_items = []
cursor = None
params = initial_params or {}
while True:
if cursor:
params['cursor'] = cursor
response = requests.get(base_url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
all_items.extend(data.get('results', []))
pagination = data.get('pagination', {})
cursor = pagination.get('cursor')
if not cursor:
break
return all_items
Build an abstraction layer that handles all of these, or you'll be writing the same pagination logic everywhere with subtle bugs in each implementation:
from abc import ABC, abstractmethod
from typing import Iterator, Any
class Paginator(ABC):
"""Abstract base class for different pagination strategies."""
@abstractmethod
def paginate(self, url: str, **kwargs) -> Iterator[dict]:
"""Yield pages of results."""
pass
class GraphPaginator(Paginator):
def __init__(self, client: GraphClient):
self.client = client
def paginate(self, url: str, **kwargs) -> Iterator[dict]:
next_url = url
while next_url:
response = self.client.request(next_url).json()
yield response
next_url = response.get('@odata.nextLink')
class CursorPaginator(Paginator):
def paginate(self, url: str, **kwargs) -> Iterator[dict]:
cursor = None
while True:
params = kwargs.copy()
if cursor:
params['cursor'] = cursor
response = requests.get(url, params=params, timeout=30).json()
yield response
cursor = response.get('pagination', {}).get('cursor')
if not cursor:
break
# Usage
graph_paginator = GraphPaginator(graph_client)
for page in graph_paginator.paginate('https://graph.microsoft.com/v1.0/users'):
users = page.get('value', [])
process_users(users)
4. Beta APIs: Production Code Built on Quicksand
Microsoft Graph's beta endpoints are where features go to die — or randomly change on Tuesday with no warning. It's like building your house on the set of Total Recall — everything looks real until the air runs out.
We had no choice. The features we needed only existed in beta. So we learned to live dangerously:
Version everything explicitly:
class GraphEndpoints:
"""Explicit API version management."""
BETA = 'https://graph.microsoft.com/beta'
V1 = 'https://graph.microsoft.com/v1.0'
@classmethod
def users_v1(cls) -> str:
return f"{cls.V1}/users"
@classmethod
def users_beta(cls) -> str:
return f"{cls.BETA}/users"
@classmethod
def sensitivity_labels_beta(cls) -> str:
"""Beta-only endpoint - may break at any time."""
return f"{cls.BETA}/me/informationProtection/policy/labels"
# Never use generic URLs
# BAD: 'https://graph.microsoft.com/users'
# GOOD: GraphEndpoints.users_v1()
Expect response schema changes. Add defensive parsing like you're defusing a bomb in Lethal Weapon:
from typing import Optional
def parse_user_profile(response: dict) -> dict:
"""Parse user profile with fallbacks for schema changes."""
return {
'id': response.get('id') or response.get('userId') or response.get('objectId'),
'email': (
response.get('mail') or
response.get('userPrincipalName') or
response.get('email') or
'[email protected]'
),
'name': (
response.get('displayName') or
response.get('name') or
response.get('fullName') or
'Unknown User'
),
'department': response.get('department', 'N/A')
}
def safe_get_nested(data: dict, *keys, default=None):
"""Safely navigate nested dicts."""
for key in keys:
if isinstance(data, dict):
data = data.get(key)
if data is None:
return default
else:
return default
return data
# Usage
user_data = response.json()
manager_email = safe_get_nested(user_data, 'manager', 'mail', default='[email protected]')
Catch and log everything. When beta APIs break (not if, when), you need evidence. Make Sentry your best friend:
import sentry_sdk
import logging
logger = logging.getLogger(__name__)
def call_beta_api(url: str, client: GraphClient) -> dict:
"""Call beta API with comprehensive error tracking."""
try:
response = client.request(url)
response.raise_for_status()
return response.json()
except Exception as e:
# Log everything
logger.error(
"Graph beta API failure",
extra={
'url': url,
'status': getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None,
'response_body': getattr(e.response, 'text', None) if hasattr(e, 'response') else None,
'timestamp': time.time()
}
)
# Send to Sentry with context
with sentry_sdk.push_scope() as scope:
scope.set_tag('api', 'graph-beta')
scope.set_tag('endpoint', url)
scope.set_context('response', {
'status': getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None,
'body': getattr(e.response, 'text', None) if hasattr(e, 'response') else None
})
sentry_sdk.capture_exception(e)
raise
The hardest lesson: When a beta API breaks, Microsoft support will shrug and say "beta isn't supported." You can't point customers to official docs saying it's broken because there are no docs. It's like trying to convince Deckard that Rachael is a replicant — you know the truth, but you can't prove it.
Document your own API behavior. Keep screenshots. Save successful responses as test fixtures:
import json
from pathlib import Path
from datetime import datetime
class BetaAPIDocumenter:
"""Document beta API responses for future reference."""
def __init__(self, storage_path: str = './api_docs'):
self.storage_path = Path(storage_path)
self.storage_path.mkdir(exist_ok=True)
def save_response(self, endpoint: str, response: dict):
"""Save successful response as evidence."""
timestamp = datetime.now().isoformat()
safe_endpoint = endpoint.replace('/', '_').replace(':', '')
filename = f"{safe_endpoint}_{timestamp}.json"
filepath = self.storage_path / filename
with open(filepath, 'w') as f:
json.dump({
'endpoint': endpoint,
'timestamp': timestamp,
'response': response
}, f, indent=2)
def load_latest(self, endpoint: str) -> Optional[dict]:
"""Load most recent documented response."""
safe_endpoint = endpoint.replace('/', '_').replace(':', '')
files = sorted(self.storage_path.glob(f"{safe_endpoint}_*.json"), reverse=True)
if files:
with open(files[0]) as f:
return json.load(f)
return None
# Usage
documenter = BetaAPIDocumenter()
# When API works, save it
response = call_beta_api(GraphEndpoints.sensitivity_labels_beta(), client)
documenter.save_response('sensitivity_labels', response)
# When API breaks, show evidence
last_working = documenter.load_latest('sensitivity_labels')
print(f"Last time this worked: {last_working['timestamp']}")
5. Reverse Engineering: When You Need What They Won't Give You
Sometimes the API you need doesn't exist. Or it does, but it's not documented. Or it's documented but requires enterprise licensing that costs more than your entire startup budget. Welcome to the Escape from New York scenario — you're going in without a map.
We needed to read sensitivity labels from Microsoft Information Protection (MIP). The official SDK did it, but only in desktop apps with user interaction. We needed it server-side, programmatically. The official API didn't exist.
Step 1: Intercept SDK requests
Run the official SDK through a proxy (mitmproxy is your friend):
# Install mitmproxy
pip install mitmproxy
# Start proxy
mitmproxy -p 8888
# Set environment variables for your test script
export HTTP_PROXY=http://localhost:8888
export HTTPS_PROXY=http://localhost:8888
export REQUESTS_CA_BUNDLE=~/.mitmproxy/mitmproxy-ca-cert.pem
# Run the official SDK example
python official_sdk_example.py
Watch what requests it makes. Copy the endpoint URL, headers, and request body. Take screenshots. This is your evidence.
Step 2: Figure out authentication
The SDK used a specific OAuth scope we didn't know existed:
# What we discovered from intercepting requests
HIDDEN_SCOPES = [
'https://graph.microsoft.com/.default',
'https://graph.microsoft.com/InformationProtectionPolicy.Read.All', # Undocumented!
]
def get_token_for_hidden_api(tenant_id: str, client_id: str, client_secret: str) -> str:
"""Get token with undocumented scopes."""
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'scope': ' '.join(HIDDEN_SCOPES)
}
response = requests.post(token_url, data=data, timeout=30)
response.raise_for_status()
return response.json()['access_token']
Step 3: Replicate the request
def get_sensitivity_labels_undocumented(token: str) -> list[dict]:
"""
Call undocumented MIP API.
WARNING: This is reverse-engineered and may break at any time.
No official documentation exists.
Microsoft support will not help if this breaks.
Use at your own risk.
"""
url = 'https://graph.microsoft.com/beta/me/informationProtection/policy/labels'
# These headers were discovered from SDK traffic
headers = {
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json',
'x-client-SKU': 'python', # From intercepted requests
'x-client-version': '1.0.0',
'User-Agent': 'CustomMIPClient/1.0'
}
try:
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
return response.json().get('value', [])
except Exception as e:
logger.error(
"Undocumented API failed (expected)",
extra={'endpoint': url, 'error': str(e)}
)
# Send alert - this is high-risk code
sentry_sdk.capture_exception(e, level='warning')
raise
Step 4: Accept the risks and document everything
class UndocumentedAPIClient:
"""
Client for reverse-engineered APIs.
RISKS:
- No documentation means no guarantees
- Endpoint can change without notice
- Breaking changes won't be announced
- Microsoft support will not help
- May violate ToS (review carefully)
MITIGATION:
- Extensive logging and monitoring
- Fallback mechanisms required
- Document all reverse-engineering steps
- Keep evidence of successful calls
- Version lock everything
"""
def __init__(self, tenant_id: str, client_id: str, client_secret: str):
self.tenant_id = tenant_id
self.client_id = client_id
self.client_secret = client_secret
self.documenter = BetaAPIDocumenter('./undocumented_apis')
def call_with_monitoring(self, endpoint: str, **kwargs) -> dict:
"""Call undocumented API with full monitoring."""
try:
token = get_token_for_hidden_api(
self.tenant_id,
self.client_id,
self.client_secret
)
# Make the call
response = get_sensitivity_labels_undocumented(token)
# Document success
self.documenter.save_response(endpoint, response)
# Alert on schema changes
last_response = self.documenter.load_latest(endpoint)
if last_response and not self._schemas_match(response, last_response['response']):
logger.warning(f"Schema change detected in {endpoint}")
sentry_sdk.capture_message(
f"Undocumented API schema changed: {endpoint}",
level='warning'
)
return response
except Exception as e:
logger.error(f"Undocumented API call failed: {endpoint}")
raise
def _schemas_match(self, new: dict, old: dict) -> bool:
"""Check if response schemas match (basic check)."""
if isinstance(new, list) and isinstance(old, list):
if len(new) > 0 and len(old) > 0:
return set(new[0].keys()) == set(old[0].keys())
return True
We documented our reverse-engineered endpoints extensively, monitored them closely in Sentry, and had fallback plans. Worth it? For our use case, yes. Calculate your own risk. It's like stealing the plans from the Empire—sometimes you just have to do it.
6. Email APIs: Where Hope Goes to Die
Email delivery APIs like Mailgun and AWS SES are deceptively simple. Send email, get 200 OK, done. Except it's never done. It's like Groundhog Day, but instead of Bill Murray you get bounce notifications.
Success doesn't mean delivery. The API says "200 OK." The email bounces 5 minutes later. Or gets marked as spam. Or lands in a mysterious void where emails go to contemplate their existence.
Webhooks are mandatory:
from flask import Flask, request
import hmac
import hashlib
app = Flask(__name__)
@app.route('/webhooks/ses', methods=['POST'])
def ses_webhook():
"""Handle AWS SNS notifications for SES events."""
# Verify SNS signature (always verify webhooks!)
message = request.json
if message.get('Type') == 'SubscriptionConfirmation':
# Confirm SNS subscription
requests.get(message['SubscribeURL'])
return 'OK', 200
if message.get('Type') == 'Notification':
ses_message = json.loads(message['Message'])
notification_type = ses_message.get('notificationType')
if notification_type == 'Bounce':
handle_bounce(ses_message['bounce'])
elif notification_type == 'Complaint':
handle_complaint(ses_message['complaint'])
elif notification_type == 'Delivery':
handle_delivery(ses_message['delivery'])
return 'OK', 200
def handle_bounce(bounce: dict):
"""Handle email bounces."""
bounce_type = bounce['bounceType']
recipients = bounce['bouncedRecipients']
for recipient in recipients:
email = recipient['emailAddress']
if bounce_type == 'Permanent':
# Hard bounce - remove from list
logger.warning(f"Permanent bounce for {email}")
mark_email_invalid(email)
elif bounce_type == 'Transient':
# Soft bounce - retry later
logger.info(f"Transient bounce for {email}")
schedule_retry(email, delay_minutes=60)
def handle_complaint(complaint: dict):
"""Handle spam complaints."""
recipients = complaint['complainedRecipients']
for recipient in recipients:
email = recipient['emailAddress']
logger.warning(f"Spam complaint from {email}")
# Immediately unsubscribe
unsubscribe_email(email)
# Track complaint rate
increment_complaint_counter()
def mark_email_invalid(email: str):
"""Mark email as permanently invalid."""
# Update database
db.execute(
"UPDATE users SET email_valid = FALSE, bounced_at = NOW() WHERE email = %s",
(email,)
)
def schedule_retry(email: str, delay_minutes: int):
"""Schedule email retry after transient failure."""
# Add to retry queue
retry_queue.enqueue_in(
timedelta(minutes=delay_minutes),
'send_email',
email=email
)
Provider-specific reputation matters. An email might bounce from Gmail with "bad reputation" but deliver fine to Outlook. Your domain reputation is per-provider. Track it:
def analyze_bounce_rates():
"""Analyze bounce rates by email provider."""
query = """
SELECT
SUBSTRING(email FROM POSITION('@' IN email) + 1) as domain,
COUNT(*) as total_sent,
SUM(CASE WHEN status = 'bounced' THEN 1 ELSE 0 END) as bounced,
ROUND(100.0 * SUM(CASE WHEN status = 'bounced' THEN 1 ELSE 0 END) / COUNT(*), 2) as bounce_rate
FROM email_logs
WHERE sent_at > NOW() - INTERVAL '7 days'
GROUP BY domain
HAVING COUNT(*) > 100
ORDER BY bounce_rate DESC
"""
results = db.execute(query)
for row in results:
if row['bounce_rate'] > 10:
logger.warning(
f"High bounce rate for {row['domain']}: {row['bounce_rate']}%"
)
# Alert if Gmail specifically has issues
if 'gmail.com' in row['domain']:
sentry_sdk.capture_message(
f"Gmail reputation issue: {row['bounce_rate']}% bounce rate",
level='warning'
)
Classify every error:
from enum import Enum
class BounceType(Enum):
HARD = 'hard' # Permanent - remove from list
SOFT = 'soft' # Transient - retry later
BLOCK = 'block' # Policy - fix configuration
COMPLAINT = 'complaint' # Spam - unsubscribe immediately
def classify_bounce(bounce_message: str, smtp_code: str) -> BounceType:
"""Classify bounce type from error message."""
message_lower = bounce_message.lower()
# Hard bounces (permanent)
hard_bounce_indicators = [
'does not exist',
'unknown user',
'invalid recipient',
'user unknown',
'no such user',
'recipient address rejected'
]
if any(indicator in message_lower for indicator in hard_bounce_indicators):
return BounceType.HARD
# Soft bounces (transient)
soft_bounce_indicators = [
'mailbox full',
'quota exceeded',
'timeout',
'temporarily unavailable',
'try again later'
]
if any(indicator in message_lower for indicator in soft_bounce_indicators):
return BounceType.SOFT
# Policy blocks
block_indicators = [
'spam',
'blacklist',
'reputation',
'policy',
'dmarc',
'spf',
'dkim'
]
if any(indicator in message_lower for indicator in block_indicators):
return BounceType.BLOCK
# Default to soft bounce if unsure
return BounceType.SOFT
Build idempotency into everything:
import uuid
from datetime import datetime
def send_email_idempotent(
to: str,
subject: str,
body: str,
message_id: str = None
) -> dict:
"""Send email with idempotency."""
# Generate or use provided message ID
if not message_id:
message_id = str(uuid.uuid4())
# Check if already sent
existing = db.execute(
"SELECT * FROM email_logs WHERE message_id = %s",
(message_id,)
).fetchone()
if existing and existing['status'] == 'delivered':
logger.info(f"Email {message_id} already delivered, skipping")
return existing
# Send via SES
try:
ses = boto3.client('ses')
response = ses.send_email(
Source='[email protected]',
Destination={'ToAddresses': [to]},
Message={
'Subject': {'Data': subject},
'Body': {'Html': {'Data': body}}
}
)
# Store result
db.execute("""
INSERT INTO email_logs (message_id, recipient, status, provider_message_id, sent_at)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (message_id) DO UPDATE
SET status = EXCLUDED.status, provider_message_id = EXCLUDED.provider_message_id
""", (
message_id,
to,
'sent',
response['MessageId'],
datetime.now()
))
return {
'message_id': message_id,
'provider_message_id': response['MessageId'],
'status': 'sent'
}
except Exception as e:
logger.error(f"Failed to send email {message_id}: {e}")
db.execute("""
INSERT INTO email_logs (message_id, recipient, status, error_message, sent_at)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (message_id) DO UPDATE
SET status = EXCLUDED.status, error_message = EXCLUDED.error_message
""", (
message_id,
to,
'failed',
str(e),
datetime.now()
))
raise
Webhook might arrive before API response returns. Or fire twice. Or arrive 10 minutes later. Idempotency saves you from sending duplicate emails and looking like an amateur.
The Meta-Lesson: Trust Nothing, Verify Everything
After years of this, here's what I've learned. Think of it as the wisdom Yoda would share if he'd spent 25 years debugging API integrations instead of training Jedi:
1. Fail fast and loud. Don't silently swallow errors hoping they'll go away. They won't. They'll multiply like gremlins after midnight:
import sentry_sdk
import logging
# Initialize Sentry
sentry_sdk.init(
dsn="your-dsn-here",
traces_sample_rate=0.1,
environment="production"
)
logger = logging.getLogger(__name__)
def api_call_with_monitoring(url: str) -> dict:
"""API call with comprehensive monitoring."""
try:
response = requests.get(url, timeout=30)
response.raise_for_status()
return response.json()
except Exception as e:
# Log with context
logger.error(
"API call failed",
extra={
'url': url,
'status': getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None,
'error': str(e)
},
exc_info=True
)
# Send to Sentry
with sentry_sdk.push_scope() as scope:
scope.set_tag('api_url', url)
scope.set_context('response', {
'status': getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None,
'body': getattr(e.response, 'text', None) if hasattr(e, 'response') else None
})
sentry_sdk.capture_exception(e)
raise
2. Build circuit breakers. If an API fails 10 times in a row, stop hammering it. You're not Rocky Balboa — you can't just keep getting hit:
from enum import Enum
from datetime import datetime, timedelta
class CircuitState(Enum):
CLOSED = 'closed' # Normal operation
OPEN = 'open' # Failing, reject requests
HALF_OPEN = 'half_open' # Testing if recovered
class CircuitBreaker:
"""Circuit breaker pattern for external APIs."""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time = None
self.state = CircuitState.CLOSED
def call(self, func, *args, **kwargs):
"""Execute function with circuit breaker protection."""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise Exception(
f"Circuit breaker is OPEN. "
f"Next attempt at {self.last_failure_time + timedelta(seconds=self.recovery_timeout)}"
)
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""Check if enough time has passed to try again."""
return (
self.last_failure_time and
datetime.now() >= self.last_failure_time + timedelta(seconds=self.recovery_timeout)
)
def _on_success(self):
"""Reset on successful call."""
self.failure_count = 0
self.state = CircuitState.CLOSED
def _on_failure(self):
"""Handle failure."""
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(
f"Circuit breaker opened after {self.failure_count} failures"
)
# Usage
graph_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
def get_user_with_breaker(user_id: str) -> dict:
return graph_breaker.call(
api_call_with_monitoring,
f"https://graph.microsoft.com/v1.0/users/{user_id}"
)
3. Monitor external API health separately. Don't let external API failures pollute your application error rates:
class ExternalAPIMetrics:
"""Track external API health separately from app health."""
def __init__(self):
self.metrics = {}
def record_call(
self,
provider: str,
endpoint: str,
status: int,
duration_ms: float,
error: str = None
):
"""Record API call metrics."""
key = f"{provider}:{endpoint}"
if key not in self.metrics:
self.metrics[key] = {
'total_calls': 0,
'failed_calls': 0,
'avg_duration_ms': 0,
'errors': []
}
self.metrics[key]['total_calls'] += 1
if status >= 400 or error:
self.metrics[key]['failed_calls'] += 1
if error and error not in self.metrics[key]['errors']:
self.metrics[key]['errors'].append(error)
# Update rolling average duration
current_avg = self.metrics[key]['avg_duration_ms']
total = self.metrics[key]['total_calls']
self.metrics[key]['avg_duration_ms'] = (
(current_avg * (total - 1) + duration_ms) / total
)
# Log to monitoring system (Datadog, CloudWatch, etc.)
logger.info(
"External API call",
extra={
'category': 'external_api', # Separate from app errors
'provider': provider,
'endpoint': endpoint,
'status': status,
'duration_ms': duration_ms,
'error': error
}
)
# Usage
metrics = ExternalAPIMetrics()
import time
def monitored_api_call(url: str) -> dict:
"""API call with metrics tracking."""
start = time.time()
error = None
status = None
try:
response = requests.get(url, timeout=30)
status = response.status_code
response.raise_for_status()
return response.json()
except Exception as e:
error = str(e)
status = getattr(e.response, 'status_code', 0) if hasattr(e, 'response') else 0
raise
finally:
duration_ms = (time.time() - start) * 1000
metrics.record_call(
provider='microsoft_graph',
endpoint=url.split('graph.microsoft.com')[1] if 'graph.microsoft.com' in url else url,
status=status or 0,
duration_ms=duration_ms,
error=error
)
4. Have fallbacks. Can't reach Microsoft Graph? Maybe you cached user data 10 minutes ago. Use it. Stale data beats no data:
import redis
import pickle
from typing import Optional
class CachedAPIClient:
"""API client with fallback to cache."""
def __init__(self, redis_client: redis.Redis, cache_ttl: int = 600):
self.redis = redis_client
self.cache_ttl = cache_ttl
def get_with_fallback(self, key: str, fetch_func, *args, **kwargs) -> dict:
"""Try API, fallback to cache if fails."""
cache_key = f"cache:{key}"
try:
# Try fresh data
data = fetch_func(*args, **kwargs)
# Cache success
self.redis.setex(
cache_key,
self.cache_ttl,
pickle.dumps(data)
)
return data
except Exception as e:
logger.warning(f"API call failed, trying cache: {e}")
# Try cache
cached = self.redis.get(cache_key)
if cached:
logger.info(f"Serving stale data for {key}")
return pickle.loads(cached)
# No cache, fail
logger.error(f"No cached data available for {key}")
raise
# Usage
cached_client = CachedAPIClient(redis.Redis())
def get_user_safe(user_id: str) -> dict:
return cached_client.get_with_fallback(
key=f"user:{user_id}",
fetch_func=lambda: api_call_with_monitoring(
f"https://graph.microsoft.com/v1.0/users/{user_id}"
)
)
5. Build test environments that break. Mock your external APIs to return errors. If your code can't handle a 429 in testing, it won't handle it in production:
from unittest.mock import Mock, patch
import pytest
class MockGraphAPI:
"""Mock Microsoft Graph that simulates failures."""
def __init__(self):
self.call_count = 0
self.should_throttle_on = [3, 6, 9] # Throttle on these call numbers
self.should_500_on = [5]
def get_user(self, user_id: str) -> dict:
"""Simulate Graph API with occasional failures."""
self.call_count += 1
# Simulate throttling
if self.call_count in self.should_throttle_on:
response = Mock()
response.status_code = 429
response.headers = {'Retry-After': '2'}
raise requests.HTTPError(response=response)
# Simulate server error
if self.call_count in self.should_500_on:
response = Mock()
response.status_code = 500
raise requests.HTTPError(response=response)
# Success
return {'id': user_id, 'name': 'Test User'}
def test_handles_throttling():
"""Test that our code handles 429 correctly."""
mock_api = MockGraphAPI()
# Should retry and eventually succeed
with patch('requests.get', side_effect=mock_api.get_user):
result = get_user_with_retry('test-user')
assert result['id'] == 'test-user'
assert mock_api.call_count >= 3 # Had to retry
6. Read the provider's status page religiously. Subscribe to status notifications. When things break, knowing "AWS S3 is down in us-east-1" saves you hours of debugging:
import feedparser
from datetime import datetime
def check_provider_status(provider: str) -> list[dict]:
"""Check provider status page."""
status_feeds = {
'aws': 'https://status.aws.amazon.com/rss/all.rss',
'microsoft': 'https://status.cloud.microsoft/en-us/status/feed/',
'github': 'https://www.githubstatus.com/history.rss'
}
if provider not in status_feeds:
return []
feed = feedparser.parse(status_feeds[provider])
recent_incidents = []
for entry in feed.entries[:5]: # Last 5 incidents
recent_incidents.append({
'title': entry.title,
'link': entry.link,
'published': entry.published,
'summary': entry.summary
})
return recent_incidents
# Check before escalating
incidents = check_provider_status('microsoft')
if incidents:
logger.info(f"Recent Microsoft incidents: {incidents}")
7. Version your integrations. When you make changes to how you call an external API, tag it:
class VersionedGraphClient:
"""Graph client with integration version tracking."""
INTEGRATION_VERSION = "2.1.0" # Bump when changing API integration
def request(self, url: str) -> requests.Response:
"""Make request with version header."""
headers = {
'Authorization': f'Bearer {self.get_token()}',
'X-Integration-Version': self.INTEGRATION_VERSION,
'User-Agent': f'MyApp/{self.INTEGRATION_VERSION}'
}
response = requests.get(url, headers=headers, timeout=30)
# Log version with every call
logger.info(
"Graph API call",
extra={
'integration_version': self.INTEGRATION_VERSION,
'url': url,
'status': response.status_code
}
)
return response
The Bottom Line
External APIs are amazing. They let you build features in days that would take months from scratch. Microsoft Graph gives you access to the entire Office 365 universe. AWS SES sends emails at pennies per thousand. These are powerful tools.
But they're also landmines. Microsoft will throttle you without warning. AWS will have a bad day and take your service with it. Beta APIs will change and break your production code at 2 AM on a Saturday.
Build defensively. Retry intelligently. Monitor obsessively. Have circuit breakers, timeouts, and fallbacks. Classify errors. Test failure scenarios. Document the undocumented. Cache aggressively. Version everything.
And when something inevitably breaks at 3 AM, you'll be grateful you did. You'll also be grateful you watched Die Hard because that's the energy you need — one man against an impossible system, armed only with clever workarounds and a sense of humor.
Now if you'll excuse me, I need to go investigate why our AWS SES reputation score just dropped. Again. It's probably Gmail. It's always Gmail.
Got your own external API war stories? Think I missed a critical pattern? Drop me a comment on LinkedIn or reach out via blog.hermesc.gr. Let's share the pain and maybe save someone else from learning these lessons the hard way.
Member discussion: