APIs change without warning. Fields get renamed, endpoints get deprecated, rate limits shift. If your application depends on third-party APIs, you need automated monitoring to catch breaking changes before your users do.
Why API Monitoring Matters
A 2024 study found that 60% of API integrations break at least once per year due to undocumented changes. Most teams only discover the break when users report errors. Automated API monitoring catches these changes within minutes.
Architecture
Our monitoring system:
- Stores API contracts (expected responses)
- Runs periodic checks against live endpoints
- Diffs responses against contracts
- Alerts on breaking changes
Setting Up
pip install requests deepdiff jsonschema pyyaml schedule
Defining API Contracts
Store expected API behavior in YAML:
# contracts/github_api.yaml
name: GitHub API
base_url: https://api.github.com
endpoints:
- path: /users/octocat
method: GET
expected_status: 200
required_fields:
- login
- id
- avatar_url
- type
field_types:
login: str
id: int
public_repos: int
max_response_time: 2.0
- path: /repos/python/cpython
method: GET
expected_status: 200
required_fields:
- full_name
- description
- stargazers_count
The Contract Checker
import requests
import yaml
import time
from deepdiff import DeepDiff
from datetime import datetime
class APIMonitor:
def __init__(self, contract_path):
with open(contract_path) as f:
self.contract = yaml.safe_load(f)
self.base_url = self.contract["base_url"]
self.results = []
def check_endpoint(self, endpoint):
"""Test a single endpoint against its contract."""
url = f"{self.base_url}{endpoint['path']}"
issues = []
start = time.time()
try:
resp = requests.get(url, timeout=10)
elapsed = time.time() - start
except requests.RequestException as e:
return [{"severity": "critical", "issue": f"Request failed: {e}"}]
# Check status code
expected_status = endpoint.get("expected_status", 200)
if resp.status_code != expected_status:
issues.append({
"severity": "critical",
"issue": f"Status {resp.status_code}, expected {expected_status}"
})
return issues
# Check response time
max_time = endpoint.get("max_response_time", 5.0)
if elapsed > max_time:
issues.append({
"severity": "warning",
"issue": f"Response time {elapsed:.2f}s > {max_time}s"
})
# Check required fields
try:
data = resp.json()
except ValueError:
issues.append({"severity": "critical", "issue": "Response is not valid JSON"})
return issues
for field in endpoint.get("required_fields", []):
if field not in data:
issues.append({
"severity": "critical",
"issue": f"Missing required field: {field}"
})
# Check field types
for field, expected_type in endpoint.get("field_types", {}).items():
if field in data:
actual_type = type(data[field]).__name__
if actual_type != expected_type:
issues.append({
"severity": "warning",
"issue": f"Field '{field}' type changed: {expected_type} -> {actual_type}"
})
return issues
def run_all_checks(self):
"""Check all endpoints in the contract."""
print(f"\nChecking {self.contract['name']}...")
all_issues = {}
for endpoint in self.contract["endpoints"]:
path = endpoint["path"]
issues = self.check_endpoint(endpoint)
if issues:
all_issues[path] = issues
for issue in issues:
icon = "X" if issue["severity"] == "critical" else "!"
print(f" [{icon}] {path}: {issue['issue']}")
else:
print(f" [OK] {path}")
return all_issues
Response Schema Diffing
Detect structural changes by comparing against a saved baseline:
import json
from pathlib import Path
class SchemaDiffer:
def __init__(self, baseline_dir="baselines"):
self.baseline_dir = Path(baseline_dir)
self.baseline_dir.mkdir(exist_ok=True)
def save_baseline(self, endpoint_key, response_data):
"""Save current response as baseline."""
path = self.baseline_dir / f"{endpoint_key}.json"
with open(path, "w") as f:
json.dump(response_data, f, indent=2)
def diff_against_baseline(self, endpoint_key, current_data):
"""Compare current response against saved baseline."""
path = self.baseline_dir / f"{endpoint_key}.json"
if not path.exists():
self.save_baseline(endpoint_key, current_data)
return None
with open(path) as f:
baseline = json.load(f)
diff = DeepDiff(
baseline, current_data,
ignore_order=True,
exclude_paths=[
"root['updated_at']",
"root['pushed_at']"
]
)
changes = []
if "dictionary_item_added" in diff:
changes.append(f"New fields: {diff['dictionary_item_added']}")
if "dictionary_item_removed" in diff:
changes.append(f"Removed fields: {diff['dictionary_item_removed']}")
if "type_changes" in diff:
changes.append(f"Type changes: {diff['type_changes']}")
return changes if changes else None
Setting Up Scheduled Monitoring
import schedule
def run_monitoring():
"""Run all API contract checks."""
contracts = Path("contracts").glob("*.yaml")
all_issues = {}
for contract_path in contracts:
monitor = APIMonitor(str(contract_path))
issues = monitor.run_all_checks()
if issues:
all_issues[contract_path.stem] = issues
if all_issues:
send_alert(all_issues)
return all_issues
def send_alert(issues):
"""Send alert for detected API changes."""
message = "API Changes Detected:\n\n"
for api, endpoints in issues.items():
message += f"## {api}\n"
for path, path_issues in endpoints.items():
for issue in path_issues:
message += f" - [{issue['severity']}] {path}: {issue['issue']}\n"
# Send via webhook, email, or Slack
print(message)
# Run every 15 minutes
schedule.every(15).minutes.do(run_monitoring)
Monitoring Third-Party APIs at Scale
When monitoring dozens of APIs, you need reliable infrastructure. Use ScraperAPI when endpoints require browser rendering or IP rotation. For monitoring APIs behind geographic restrictions, ThorData provides proxies in 195+ countries. Track your monitoring pipeline health with ScrapeOps.
Historical Change Tracking
def log_change(api_name, endpoint, change_type, details):
"""Log API changes to a database for trend analysis."""
conn = sqlite3.connect("api_changes.db")
conn.execute("""
CREATE TABLE IF NOT EXISTS changes (
id INTEGER PRIMARY KEY,
timestamp TEXT, api TEXT,
endpoint TEXT, change_type TEXT,
details TEXT
)
""")
conn.execute(
"INSERT INTO changes VALUES (NULL, ?, ?, ?, ?, ?)",
(datetime.now().isoformat(), api_name,
endpoint, change_type, details)
)
conn.commit()
conn.close()
Conclusion
API monitoring is insurance for your integrations. The initial setup takes an afternoon, but it saves days of debugging when APIs change unexpectedly. Start with your most critical API dependencies, define contracts, and set up 15-minute checks. When a breaking change hits, you'll know before your users do.
Top comments (0)