Python Examples
Complete code examples for integrating Status Check with Python applications.
Installation
pip install requests
Basic Setup
import requests
from typing import Dict, List, Optional
# Configuration
API_KEY = "sk_live_your_api_key_here"
BASE_URL = "https://api.status-check.io"
# Headers for all requests
HEADERS = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
Single Lead Validation
Create and Validate a Lead
def validate_lead(email: str, website: str) -> Dict:
"""
Create and validate a single lead.
Args:
email: Lead's email address
website: Lead's website URL
Returns:
Lead validation results
Raises:
Exception: If insufficient credits or API error
"""
payload = {
"email": email,
"website": website,
"validate": true,
"validateDomain": True,
"validateEmail": True
}
response = requests.post(
f"{BASE_URL}/v1/leads",
headers=HEADERS,
json=payload
)
if response.status_code == 201:
return response.json()
elif response.status_code == 402:
raise Exception("Insufficient credits. Please add credits to continue.")
else:
response.raise_for_status()
# Usage
try:
result = validate_lead(
email="john@example.com",
website="https://example.com"
)
print(f"Lead ID: {result['leadId']}")
print(f"Deliverability Score: {result['leadDeliverabilityRating']}/100")
print(f"Email Valid: {result['emailValid']}")
print(f"Domain Status: {result['domainHttpStatus']}")
print(f"Recommendation: {result['leadRecommendation']}")
except Exception as e:
print(f"Error: {e}")
Get Lead Details
def get_lead(lead_id: str) -> Dict:
"""Get details for a specific lead."""
response = requests.get(
f"{BASE_URL}/v1/leads/{lead_id}",
headers=HEADERS
)
response.raise_for_status()
return response.json()
# Usage
lead = get_lead("lead_abc123")
print(f"Email: {lead['email']}")
print(f"Score: {lead['leadDeliverabilityRating']}")
Bulk Lead Processing
Validate Multiple Leads
def bulk_validate_leads(leads: List[Dict]) -> Dict:
"""
Validate multiple leads at once.
Args:
leads: List of lead dictionaries with email and website
Returns:
Batch validation results
"""
payload = {
"leads": leads,
"validateDomain": True,
"validateEmail": True
}
response = requests.post(
f"{BASE_URL}/v1/leads/bulk",
headers=HEADERS,
json=payload
)
response.raise_for_status()
return response.json()
# Usage
leads = [
{
"email": "alice@company.com",
"website": "https://company.com",
"firstName": "Alice",
"lastName": "Smith"
},
{
"email": "bob@startup.io",
"website": "https://startup.io",
"firstName": "Bob",
"lastName": "Jones"
}
]
batch = bulk_validate_leads(leads)
print(f"Created: {batch['created']} leads")
print(f"Skipped: {batch['skipped']} duplicates")
Create Lead with CustomFields
Use customFields for CSV imports, industry-specific data, or custom properties.
def create_lead_with_custom_data(
email: str,
website: str,
custom_data: Dict[str, Any]
) -> Dict:
"""
Create lead with custom fields for arbitrary data.
Args:
email: Lead's email
website: Lead's website
custom_data: Custom key-value pairs (max 50 fields, 1KB/value)
Returns:
Created lead with customFields
"""
payload = {
"email": email,
"website": website,
"firstName": custom_data.get("firstName", ""),
"lastName": custom_data.get("lastName", ""),
"company": custom_data.get("company", ""),
"title": custom_data.get("title", ""),
"validate": True,
"validateDomain": True,
"validateEmail": True,
"customFields": {
# Industry-specific
"industry": custom_data.get("industry"),
"vertical": custom_data.get("vertical"),
"employeeCount": custom_data.get("employeeCount"),
"annualRevenue": custom_data.get("annualRevenue"),
# Sales tracking
"leadSource": custom_data.get("leadSource"),
"leadScore": custom_data.get("leadScore"),
# Any other custom data from CSV or forms
**{k: v for k, v in custom_data.items()
if k not in ["firstName", "lastName", "company", "title",
"industry", "vertical", "employeeCount",
"annualRevenue", "leadSource", "leadScore"]}
}
}
# Remove None values from customFields
payload["customFields"] = {
k: v for k, v in payload["customFields"].items()
if v is not None
}
response = requests.post(
f"{BASE_URL}/v1/leads",
headers=HEADERS,
json=payload
)
response.raise_for_status()
return response.json()
# Usage - CSV Import Example
csv_row = {
"firstName": "John",
"lastName": "Doe",
"company": "Acme Corp",
"title": "CTO",
"industry": "SaaS",
"employeeCount": "50-100",
"annualRevenue": "$5M-$10M",
"leadSource": "Trade Show",
"customColumn1": "Custom data from CSV",
"importDate": "2025-01-15"
}
lead = create_lead_with_custom_data(
email="john@acme.com",
website="https://acme.com",
custom_data=csv_row
)
print(f"Lead created: {lead['id']}")
print(f"CustomFields: {lead['customFields']}")
Update Lead with CustomFields (Merge Behavior)
CustomFields merge with existing data instead of replacing.
def update_lead_custom_fields(
lead_id: str,
updates: Dict[str, Any],
custom_updates: Dict[str, Any] = None
) -> Dict:
"""
Update lead with optional customFields merging.
CustomFields behavior:
- Existing custom fields are preserved
- New fields are added
- Updated fields override existing values
Args:
lead_id: Lead ID to update
updates: Standard field updates
custom_updates: Custom field updates (merged with existing)
Returns:
Updated lead
"""
payload = {**updates}
if custom_updates:
payload["customFields"] = custom_updates
response = requests.put(
f"{BASE_URL}/v1/leads/{lead_id}",
headers=HEADERS,
json=payload
)
response.raise_for_status()
return response.json()
# Usage Example 1: Initial customFields
lead = update_lead_custom_fields(
lead_id="lead_abc123",
updates={"notes": "Initial contact"},
custom_updates={
"industry": "SaaS",
"employeeCount": "50-100",
"timezone": "America/Los_Angeles"
}
)
print(f"Added custom fields: {lead['customFields']}")
# Usage Example 2: Merge new customFields
lead = update_lead_custom_fields(
lead_id="lead_abc123",
updates={"notes": "Follow-up scheduled"},
custom_updates={
"industry": "FinTech", # Override existing
"lastContactDate": "2025-01-15", # Add new field
"nextFollowUp": "2025-01-22" # Add new field
}
)
# Result:
# {
# "customFields": {
# "industry": "FinTech", ← Updated
# "employeeCount": "50-100", ← Preserved
# "timezone": "America/Los_Angeles", ← Preserved
# "lastContactDate": "2025-01-15", ← Added
# "nextFollowUp": "2025-01-22" ← Added
# }
# }
print(f"Merged custom fields: {lead['customFields']}")
Query Leads with Filters
def query_leads(
min_score: Optional[int] = None,
email_valid: Optional[bool] = None,
limit: int = 50
) -> List[Dict]:
"""
Query leads with optional filters.
Args:
min_score: Minimum deliverability score (0-100)
email_valid: Filter by email validity
limit: Maximum results to return
Returns:
List of matching leads
"""
params = {"limit": limit}
if min_score is not None:
params["minDeliverabilityRating"] = min_score
if email_valid is not None:
params["emailValid"] = email_valid
response = requests.get(
f"{BASE_URL}/v1/leads",
headers=HEADERS,
params=params
)
response.raise_for_status()
return response.json()["leads"]
# Usage - Get high-quality leads
high_quality_leads = query_leads(min_score=85, email_valid=True)
print(f"Found {len(high_quality_leads)} high-quality leads")
for lead in high_quality_leads:
print(f" • {lead['email']} - Score: {lead['leadDeliverabilityRating']}")
Webhook Integration
Register a Webhook
def create_webhook(
name: str,
url: str,
events: List[str]
) -> Dict:
"""
Register a webhook to receive event notifications.
Args:
name: Friendly name for the webhook
url: Your endpoint URL
events: List of events to subscribe to
Returns:
Webhook details including secret
"""
payload = {
"name": name,
"url": url,
"events": events,
"active": True
}
response = requests.post(
f"{BASE_URL}/v1/webhooks",
headers=HEADERS,
json=payload
)
response.raise_for_status()
return response.json()
# Usage
webhook = create_webhook(
name="Production Webhook",
url="https://your-app.com/webhooks/status-check",
events=["validation.complete", "credits.low"]
)
print(f"Webhook ID: {webhook['webhook_id']}")
print(f"Secret: {webhook['secret']}") # Save this securely!
Verify Webhook Signatures
import hmac
import hashlib
def verify_webhook_signature(
payload: str,
signature: str,
secret: str
) -> bool:
"""
Verify webhook signature using HMAC-SHA256.
Args:
payload: Raw request body as string
signature: X-Webhook-Signature header value
secret: Your webhook secret (whsec_...)
Returns:
True if signature is valid, False otherwise
"""
expected = hmac.new(
secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
# Use timing-safe comparison to prevent timing attacks
return hmac.compare_digest(
f"sha256={expected}",
signature
)
Flask Webhook Handler
from flask import Flask, request, jsonify
app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret_here"
@app.route('/webhooks/status-check', methods=['POST'])
def handle_webhook():
"""Handle incoming webhooks from Status Check."""
# Get raw body and signature
payload = request.get_data(as_text=True)
signature = request.headers.get('X-Webhook-Signature')
# Verify signature
if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
return jsonify({"error": "Invalid signature"}), 401
# Parse event
event = request.json
event_type = event['event']
# Handle different event types
if event_type == 'validation.complete':
batch_id = event['batchId']
total_leads = event['totalLeads']
successful = event['successfulValidations']
print(f"✓ Validation complete for batch {batch_id}")
print(f" Results: {successful}/{total_leads} successful")
# Process results (e.g., update database, send notifications)
elif event_type == 'credits.low':
balance = event['currentBalance']
print(f"⚠ Credit balance low: {balance} credits remaining")
# Send alert email, Slack notification, etc.
return jsonify({"status": "received"}), 200
if __name__ == '__main__':
app.run(port=3000)
FastAPI Webhook Handler
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel
app = FastAPI()
WEBHOOK_SECRET = "whsec_your_secret_here"
class ValidationCompleteEvent(BaseModel):
event: str
batchId: str
userId: str
totalLeads: int
successfulValidations: int
@app.post("/webhooks/status-check")
async def handle_webhook(
request: Request,
background_tasks: BackgroundTasks
):
"""Handle incoming webhooks from Status Check."""
# Get raw body and signature
body = await request.body()
payload = body.decode('utf-8')
signature = request.headers.get('X-Webhook-Signature')
# Verify signature
if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
raise HTTPException(status_code=401, detail="Invalid signature")
# Parse event
event_data = await request.json()
# Queue background processing (respond quickly)
background_tasks.add_task(process_webhook_event, event_data)
return {"status": "received"}
async def process_webhook_event(event: dict):
"""Process webhook event asynchronously."""
event_type = event['event']
if event_type == 'validation.complete':
# Update database with validation results
batch_id = event['batchId']
# ... process results
elif event_type == 'credits.low':
# Send notification to admin
# ... send alert
pass
Advanced Examples
Batch Processing with Progress Tracking
import time
def validate_large_batch(leads: List[Dict]) -> Dict:
"""
Validate a large batch of leads and track progress.
Args:
leads: List of leads to validate
Returns:
Final validation results
"""
# Start batch validation
response = requests.post(
f"{BASE_URL}/v1/leads/validate",
headers=HEADERS,
json={
"leadIds": [lead['leadId'] for lead in leads],
"validateDomain": True,
"validateEmail": True
}
)
batch = response.json()
batch_id = batch['batchId']
print(f"Batch {batch_id} started")
# Poll for completion (or use webhooks instead)
while True:
batch_status = requests.get(
f"{BASE_URL}/v1/batches/{batch_id}",
headers=HEADERS
).json()
status = batch_status['status']
progress = batch_status.get('progress', 0)
print(f"Status: {status}, Progress: {progress}%")
if status == 'complete':
return batch_status
elif status == 'failed':
raise Exception("Batch validation failed")
time.sleep(5) # Wait 5 seconds before checking again
Export Results to CSV
import csv
from typing import List
def export_leads_to_csv(
leads: List[Dict],
filename: str = "validated_leads.csv"
):
"""
Export validated leads to CSV file.
Args:
leads: List of lead dictionaries
filename: Output CSV filename
"""
fieldnames = [
'email',
'website',
'firstName',
'lastName',
'leadDeliverabilityRating',
'emailValid',
'domainHttpStatus',
'leadRecommendation',
'validationStatus'
]
with open(filename, 'w', newline='') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for lead in leads:
# Extract only the fields we want
row = {field: lead.get(field, '') for field in fieldnames}
writer.writerow(row)
print(f"✓ Exported {len(leads)} leads to {filename}")
# Usage
leads = query_leads(limit=1000)
export_leads_to_csv(leads)
Retry Logic for API Calls
import time
from typing import Callable, Any
def retry_with_backoff(
func: Callable,
max_retries: int = 3,
initial_delay: float = 1.0
) -> Any:
"""
Retry a function with exponential backoff.
Args:
func: Function to retry
max_retries: Maximum number of retry attempts
initial_delay: Initial delay in seconds
Returns:
Function result
Raises:
Last exception if all retries fail
"""
delay = initial_delay
for attempt in range(max_retries):
try:
return func()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise # Last attempt failed, re-raise exception
# Check if it's a rate limit error
if hasattr(e, 'response') and e.response.status_code == 429:
retry_after = int(e.response.headers.get('Retry-After', delay))
print(f"Rate limited. Waiting {retry_after} seconds...")
time.sleep(retry_after)
else:
print(f"Request failed (attempt {attempt + 1}/{max_retries}). Retrying in {delay}s...")
time.sleep(delay)
delay *= 2 # Exponential backoff
# Usage
result = retry_with_backoff(
lambda: validate_lead("test@example.com", "https://example.com")
)
Error Handling
Handle All API Error Codes
def safe_api_call(func: Callable) -> Optional[Dict]:
"""
Wrapper for safe API calls with comprehensive error handling.
Args:
func: API call function to wrap
Returns:
API response or None if error
"""
try:
return func()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
if status_code == 400:
print("❌ Bad Request: Check your request data")
elif status_code == 401:
print("❌ Unauthorized: Invalid API key")
elif status_code == 402:
print("❌ Insufficient Credits: Please add credits")
elif status_code == 404:
print("❌ Not Found: Resource doesn't exist")
elif status_code == 429:
print("❌ Rate Limit Exceeded: Slow down requests")
elif status_code >= 500:
print("❌ Server Error: Try again later")
return None
except requests.exceptions.ConnectionError:
print("❌ Connection Error: Check your internet connection")
return None
except requests.exceptions.Timeout:
print("❌ Request Timeout: API took too long to respond")
return None
# Usage
result = safe_api_call(
lambda: validate_lead("test@example.com", "https://example.com")
)
if result:
print(f"Success! Score: {result['leadDeliverabilityRating']}")
else:
print("Failed to validate lead")
Complete Example: Lead Enrichment Pipeline
#!/usr/bin/env python3
"""
Complete lead enrichment pipeline example.
Reads leads from CSV, validates them, and exports results.
"""
import csv
import sys
from typing import List, Dict
def read_leads_from_csv(filename: str) -> List[Dict]:
"""Read leads from CSV file."""
leads = []
with open(filename, 'r') as f:
reader = csv.DictReader(f)
for row in reader:
leads.append({
'email': row['email'],
'website': row.get('website', ''),
'firstName': row.get('firstName', ''),
'lastName': row.get('lastName', '')
})
return leads
def main():
if len(sys.argv) != 2:
print("Usage: python enrich_leads.py input.csv")
sys.exit(1)
input_file = sys.argv[1]
# Read leads
print(f"📖 Reading leads from {input_file}...")
leads = read_leads_from_csv(input_file)
print(f" Found {len(leads)} leads")
# Validate leads
print(f"🔍 Validating {len(leads)} leads...")
result = bulk_validate_leads(leads)
print(f" Created: {result['created']}")
print(f" Skipped: {result['skipped']}")
# Wait for validation to complete (using webhook is better)
print("⏳ Waiting for validation to complete...")
time.sleep(30) # Give it time to process
# Fetch validated leads
print("📊 Fetching results...")
validated_leads = query_leads(limit=len(leads))
# Export results
output_file = input_file.replace('.csv', '_validated.csv')
export_leads_to_csv(validated_leads, output_file)
# Print summary
high_quality = len([l for l in validated_leads if l['leadDeliverabilityRating'] >= 85])
print(f"\n✅ Enrichment complete!")
print(f" Total: {len(validated_leads)} leads")
print(f" High Quality (85+): {high_quality} leads")
print(f" Output: {output_file}")
if __name__ == '__main__':
main()