Integration Patterns
1. Webhook Simulation
Regularly poll for changes:
import schedule
import time
def check_new_leads():
registrants = get_registrants(property_id='prop_123')
new_leads = filter(lambda r: r['status'] == 'new', registrants)
for lead in new_leads:
process_lead(lead)
schedule.every(5).minutes.do(check_new_leads)
2. Caching Strategy
Reduce API calls with intelligent caching:
import redis
from datetime import timedelta
def get_cached_property(property_id, ttl=3600):
cache_key = f"property:{property_id}"
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
property_data = api.get_property(property_id)
redis.setex(cache_key, ttl, json.dumps(property_data))
return property_data
3. Batch Processing
Handle multiple resources efficiently:
def process_brokerage_data(brokerage_id):
# Get all registrants
all_registrants = []
page = 1
while True:
response = get_registrants_by_brokerage(brokerage_id, page=page)
all_registrants.extend(response['data'])
if page >= response['pagination']['total_pages']:
break
page += 1
# Process in batches
batch_size = 50
for i in range(0, len(all_registrants), batch_size):
batch = all_registrants[i:i+batch_size]
bulk_update_crm(batch)
4. Error Handling
Robust error management:
def safe_api_call(endpoint, params, retries=3):
for attempt in range(retries):
try:
return make_api_request(endpoint, params)
except RateLimitError:
time.sleep(2 ** attempt)
except AuthError:
log_error('Invalid API key')
raise
except ServerError:
if attempt < retries - 1:
time.sleep(5)
else:
raise