API Integration
Learn how to integrate Heimdall APIs into your applications and production systems.
Integration Patterns
REST API Integration
All Heimdall services are REST API-based, making integration straightforward:
import requests
# Standard integration pattern
def call_heimdall_api(endpoint, headers, data):
response = requests.post(endpoint, headers=headers, json=data)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API call failed: {response.status_code}")
Authentication
# Standard headers for all Heimdall APIs
headers = {
'X-api-key': 'YOUR-API-KEY',
'X-username': 'YOUR-USERNAME',
'Content-Type': 'application/json'
}
Common Integration Scenarios
Web Application Integration
// Frontend JavaScript integration
async function analyzeText(text) {
const response = await fetch('https://read.heimdallapp.org/read/v1/api/process', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-api-key': 'YOUR-API-KEY',
'X-username': 'YOUR-USERNAME'
},
body: JSON.stringify({ text: text })
});
return await response.json();
}
Backend Service Integration
# Python Flask integration
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.route('/analyze', methods=['POST'])
def analyze_data():
data = request.get_json()
# Call Heimdall API
heimdall_response = requests.post(
'https://read.heimdallapp.org/read/v1/api/process',
headers={
'X-api-key': 'YOUR-API-KEY',
'X-username': 'YOUR-USERNAME'
},
json={'text': data['text']}
)
return jsonify(heimdall_response.json())
Mobile App Integration
// iOS Swift integration
func analyzeText(_ text: String, completion: @escaping (Result<[String: Any], Error>) -> Void) {
let url = URL(string: "https://read.heimdallapp.org/read/v1/api/process")!
var request = URLRequest(url: url)
request.httpMethod = "POST"
request.setValue("YOUR-API-KEY", forHTTPHeaderField: "X-api-key")
request.setValue("YOUR-USERNAME", forHTTPHeaderField: "X-username")
request.setValue("application/json", forHTTPHeaderField: "Content-Type")
let body = ["text": text]
request.httpBody = try? JSONSerialization.data(withJSONObject: body)
URLSession.shared.dataTask(with: request) { data, response, error in
if let error = error {
completion(.failure(error))
return
}
if let data = data {
do {
let json = try JSONSerialization.jsonObject(with: data) as? [String: Any]
completion(.success(json ?? [:]))
} catch {
completion(.failure(error))
}
}
}.resume()
}
Error Handling
Standard Error Responses
def handle_heimdall_response(response):
if response.status_code == 200:
return response.json()
elif response.status_code == 401:
raise Exception("Authentication failed - check API key")
elif response.status_code == 422:
raise Exception("Invalid request data")
elif response.status_code == 429:
raise Exception("Rate limit exceeded")
else:
raise Exception(f"API error: {response.status_code}")
Retry Logic
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise e
time.sleep(delay * (2 ** attempt)) # Exponential backoff
return None
return wrapper
return decorator
@retry_on_failure(max_retries=3)
def call_heimdall_with_retry(endpoint, headers, data):
response = requests.post(endpoint, headers=headers, json=data)
if response.status_code != 200:
raise Exception(f"API call failed: {response.status_code}")
return response.json()
Performance Optimization
Caching
import redis
import json
import hashlib
# Redis cache for API responses
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def get_cached_result(cache_key):
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
return None
def cache_result(cache_key, result, ttl=3600):
redis_client.setex(cache_key, ttl, json.dumps(result))
def call_heimdall_with_cache(endpoint, headers, data):
# Create cache key from request data
cache_key = hashlib.md5(json.dumps(data).encode()).hexdigest()
# Check cache first
cached_result = get_cached_result(cache_key)
if cached_result:
return cached_result
# Make API call
response = requests.post(endpoint, headers=headers, json=data)
if response.status_code == 200:
result = response.json()
cache_result(cache_key, result)
return result
else:
raise Exception(f"API call failed: {response.status_code}")
Batch Processing
import asyncio
import aiohttp
async def batch_analyze_texts(texts, api_key, username):
async with aiohttp.ClientSession() as session:
tasks = []
for text in texts:
task = analyze_single_text(session, text, api_key, username)
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
async def analyze_single_text(session, text, api_key, username):
url = 'https://read.heimdallapp.org/read/v1/api/process'
headers = {
'X-api-key': api_key,
'X-username': username
}
data = {'text': text}
async with session.post(url, headers=headers, json=data) as response:
if response.status == 200:
return await response.json()
else:
raise Exception(f"API call failed: {response.status}")
Security Best Practices
API Key Management
import os
from cryptography.fernet import Fernet
class SecureAPIKeyManager:
def __init__(self):
self.encryption_key = os.environ.get('ENCRYPTION_KEY')
self.cipher = Fernet(self.encryption_key.encode())
def encrypt_key(self, api_key):
return self.cipher.encrypt(api_key.encode()).decode()
def decrypt_key(self, encrypted_key):
return self.cipher.decrypt(encrypted_key.encode()).decode()
def get_api_key(self):
encrypted_key = os.environ.get('HEIMDALL_API_KEY')
return self.decrypt_key(encrypted_key)
Input Validation
def validate_text_input(text):
if not isinstance(text, str):
raise ValueError("Text must be a string")
if len(text) == 0:
raise ValueError("Text cannot be empty")
if len(text) > 10000: # Adjust based on API limits
raise ValueError("Text too long")
if '\n' in text or '"' in text:
raise ValueError("Text cannot contain line breaks or quotes")
return text
def validate_image_file(file_path):
allowed_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff']
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext not in allowed_extensions:
raise ValueError(f"Unsupported file type: {file_ext}")
if os.path.getsize(file_path) > 10 * 1024 * 1024: # 10MB limit
raise ValueError("File too large")
return True
Monitoring and Logging
Request Logging
import logging
import time
from functools import wraps
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def log_api_calls(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
logger.info(f"API call successful in {duration:.2f}s")
return result
except Exception as e:
duration = time.time() - start_time
logger.error(f"API call failed after {duration:.2f}s: {e}")
raise
return wrapper
@log_api_calls
def call_heimdall_api(endpoint, headers, data):
response = requests.post(endpoint, headers=headers, json=data)
if response.status_code != 200:
raise Exception(f"API call failed: {response.status_code}")
return response.json()
Performance Metrics
import time
from collections import defaultdict
class APIMetrics:
def __init__(self):
self.call_count = 0
self.total_time = 0
self.error_count = 0
self.response_times = []
def record_call(self, duration, success=True):
self.call_count += 1
self.total_time += duration
self.response_times.append(duration)
if not success:
self.error_count += 1
def get_stats(self):
if not self.response_times:
return {}
return {
'total_calls': self.call_count,
'success_rate': (self.call_count - self.error_count) / self.call_count,
'avg_response_time': self.total_time / self.call_count,
'min_response_time': min(self.response_times),
'max_response_time': max(self.response_times)
}
# Usage
metrics = APIMetrics()
def call_with_metrics(endpoint, headers, data):
start_time = time.time()
try:
result = call_heimdall_api(endpoint, headers, data)
duration = time.time() - start_time
metrics.record_call(duration, success=True)
return result
except Exception as e:
duration = time.time() - start_time
metrics.record_call(duration, success=False)
raise
Next Steps
Now that you understand API integration:
- Connect to Databases - Integrate with your data sources
- Monitor Performance - Track API usage and performance
- Follow Best Practices - Learn production deployment tips