Best Practices¶
Follow these best practices for optimal integration with Circuit KYC.
API Key Management¶
Store Keys Securely¶
# ❌ Bad - hardcoded key
client = CircuitKYC(api_key="sk_live_abc123")
# ✅ Good - environment variable
client = CircuitKYC(api_key=os.environ["CIRCUIT_KYC_API_KEY"])
Use Separate Keys per Environment¶
| Environment | Key Type | Purpose |
|---|---|---|
| Development | Sandbox | Local testing |
| Staging | Sandbox | Pre-production tests |
| Production | Production | Real customers |
Rotate Keys Regularly¶
- Create new API key in dashboard
- Deploy with new key
- Verify functionality
- Delete old key
Phone Number Format¶
Always use E.164 format for phone numbers:
# ❌ Bad
phone = "555-123-4567"
phone = "(555) 123-4567"
phone = "5551234567"
# ✅ Good
phone = "+15551234567" # US
phone = "+442071234567" # UK
phone = "+81312345678" # Japan
Normalize Before Sending¶
import phonenumbers
def normalize_phone(phone: str, country: str = "US") -> str:
"""Normalize phone to E.164 format."""
parsed = phonenumbers.parse(phone, country)
return phonenumbers.format_number(
parsed,
phonenumbers.PhoneNumberFormat.E164
)
# Usage
phone = normalize_phone("(555) 123-4567", "US")
# Returns: +15551234567
Rate Limiting¶
Implement Client-Side Throttling¶
from ratelimit import limits, sleep_and_retry
# 300 requests per minute
@sleep_and_retry
@limits(calls=300, period=60)
def check_eligibility(email, phone):
return client.check_eligibility(email=email, phone_number=phone)
Batch Requests When Possible¶
# ❌ Bad - sequential requests
for user in users:
result = client.check_eligibility(
email=user.email,
phone_number=user.phone
)
# ✅ Good - parallel with rate limiting
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=10) as executor:
results = list(executor.map(
lambda u: client.check_eligibility(
email=u.email,
phone_number=u.phone
),
users
))
Error Handling¶
Always Handle Errors¶
# ❌ Bad - no error handling
result = client.check_eligibility(email=email, phone_number=phone)
# ✅ Good - comprehensive error handling
try:
result = client.check_eligibility(email=email, phone_number=phone)
except AuthenticationError:
logger.error("Invalid API key")
raise
except RateLimitError as e:
logger.warning(f"Rate limited, retry after {e.retry_after}s")
time.sleep(e.retry_after)
# Retry logic
except ValidationError as e:
logger.error(f"Invalid input: {e.errors}")
raise ValueError("Invalid user data")
except APIError as e:
logger.error(f"API error: {e}, request_id: {e.request_id}")
# Retry with backoff
Log Request IDs¶
try:
result = client.check_eligibility(...)
except APIError as e:
# Include request_id for support
logger.error(
"API call failed",
extra={
"request_id": e.request_id,
"status_code": e.status_code,
"error": str(e)
}
)
Credential Management¶
Store Credentials Securely¶
# ❌ Bad - store in plaintext
db.save(credential=result.token)
# ✅ Good - encrypt before storing
encrypted = encrypt(result.token, user_encryption_key)
db.save(credential_encrypted=encrypted)
Verify Before Accepting¶
def accept_credential(token: str, public_key: str) -> bool:
"""Always verify credentials before trusting them."""
result = client.verify(token=token, public_key_pem=public_key)
if not result.valid:
logger.warning(f"Invalid credential: {result.error}")
return False
# Check expiration
payload = result.payload
if payload.get('exp') and payload['exp'] < time.time():
logger.warning("Credential expired")
return False
return True
Performance¶
Use Connection Pooling¶
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Configure session with connection pooling
session = requests.Session()
adapter = HTTPAdapter(
pool_connections=10,
pool_maxsize=10,
max_retries=Retry(total=3, backoff_factor=0.5)
)
session.mount('https://', adapter)
# Pass to client
client = CircuitKYC(
api_key=os.environ["CIRCUIT_KYC_API_KEY"],
session=session
)
Cache Eligibility Results¶
from functools import lru_cache
import hashlib
@lru_cache(maxsize=1000)
def check_eligibility_cached(email_hash: str, phone_hash: str):
"""Cache eligibility results for 5 minutes."""
return client.check_eligibility(
email=original_email,
phone_number=original_phone
)
# Use hashes as cache keys (don't cache PII)
email_hash = hashlib.sha256(email.encode()).hexdigest()
phone_hash = hashlib.sha256(phone.encode()).hexdigest()
result = check_eligibility_cached(email_hash, phone_hash)
Security¶
Validate Input¶
import re
from email_validator import validate_email
def validate_user_input(email: str, phone: str) -> tuple:
"""Validate and sanitize user input before API call."""
# Validate email
try:
valid = validate_email(email)
email = valid.email
except Exception as e:
raise ValueError(f"Invalid email: {e}")
# Validate phone (E.164 format)
if not re.match(r'^\+[1-9]\d{1,14}$', phone):
raise ValueError("Phone must be in E.164 format")
return email, phone
Don't Log Sensitive Data¶
# ❌ Bad - logging PII
logger.info(f"Checking eligibility for {email}, {phone}")
# ✅ Good - log hashes or IDs only
logger.info(
"Checking eligibility",
extra={"user_id": user.id, "request_id": request_id}
)
Monitoring¶
Track API Usage¶
from prometheus_client import Counter, Histogram
api_calls = Counter(
'circuit_kyc_api_calls_total',
'Total API calls',
['endpoint', 'status']
)
api_latency = Histogram(
'circuit_kyc_api_latency_seconds',
'API call latency',
['endpoint']
)
@api_latency.labels(endpoint='check_eligibility').time()
def check_eligibility(email, phone):
try:
result = client.check_eligibility(email=email, phone_number=phone)
api_calls.labels(endpoint='check_eligibility', status='success').inc()
return result
except Exception as e:
api_calls.labels(endpoint='check_eligibility', status='error').inc()
raise
Set Up Alerts¶
Monitor for:
- Error rate > 5%
- Latency p99 > 2s
- Rate limit errors
- Credit balance < 20%
Testing¶
Use Sandbox for Tests¶
# tests/conftest.py
import pytest
@pytest.fixture
def kyc_client():
"""Use sandbox for all tests."""
return CircuitKYC(
api_key=os.environ["CIRCUIT_KYC_SANDBOX_KEY"],
environment="sandbox"
)
Mock in Unit Tests¶
from unittest.mock import Mock, patch
def test_eligibility_flow():
with patch('circuit_kyc.CircuitKYC') as mock_client:
mock_client.return_value.check_eligibility.return_value = {
"found": True,
"tier": "Tier1"
}
# Test your code
result = your_function()
assert result.verified == True
Checklist¶
Before going to production:
- [ ] API keys stored in environment variables
- [ ] Using production API key
- [ ] Error handling implemented
- [ ] Rate limiting configured
- [ ] Phone numbers normalized to E.164
- [ ] Credentials stored encrypted
- [ ] Input validation in place
- [ ] No PII in logs
- [ ] Monitoring and alerts set up
- [ ] Tested with sandbox environment