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
# Implementation is proprietary (that IS the moat).
# Skip the build — use our ready-made Apify actor:
# see the CTA below for the link (fpr=yw6md3).
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)