Best Practices

Copy

• Copy for LLM

  Copy page as Markdown for LLMs
• View as Markdown

  Open this page as Markdown
• Open in ChatGPT

  Get insights from ChatGPT
• Open in Claude

  Get insights from Claude

# Good: Secure API key handling
import os
from requests import Session

class FreddyClient:
 def __init__(self):
 self.api_key = os.getenv('FREDDY_API_KEY')
 if not self.api_key:
 raise ValueError("FREDDY_API_KEY environment variable required")

 self.session = Session()
 self.session.headers.update({
 'X-API-Key': self.api_key,
 'Content-Type': 'application/json'
 })

 def make_request(self, endpoint, **kwargs):
 response = self.session.get(f"https://api.aitronos.com/v1/{endpoint}", **kwargs)
 response.raise_for_status()
 return response.json()

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure session with connection pooling
session = requests.Session()

# Retry strategy
retry_strategy = Retry(
 total=3,
 backoff_factor=1,
 status_forcelist=[429, 500, 502, 503, 504]
)

adapter = HTTPAdapter(
 pool_connections=10,
 pool_maxsize=20,
 max_retries=retry_strategy
)

session.mount("https://", adapter)

# Good: Batch multiple operations
def process_multiple_messages(messages):
 results = []
 for batch in chunk_list(messages, 10): # Process in batches of 10
 batch_results = []
 for message in batch:
 result = freddy_client.generate_response(message)
 batch_results.append(result)
 results.extend(batch_results)
 time.sleep(0.1) # Small delay between batches
 return results

# Bad: Individual requests without batching
def process_messages_inefficiently(messages):
 results = []
 for message in messages: # Each request individually
 result = freddy_client.generate_response(message)
 results.append(result)
 return results

import redis
import json
import hashlib

class CachedFreddyClient:
 def __init__(self):
 self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
 self.cache_ttl = 3600 # 1 hour

 def generate_response_cached(self, prompt, model="gpt-4o"):
 # Create cache key from prompt and model
 cache_key = hashlib.md5(f"{prompt}:{model}".encode()).hexdigest()

 # Check cache first
 cached_result = self.redis_client.get(cache_key)
 if cached_result:
 return json.loads(cached_result)

 # Make API request if not cached
 result = self.freddy_client.generate_response(prompt, model)

 # Cache the result
 self.redis_client.setex(
 cache_key,
 self.cache_ttl,
 json.dumps(result)
 )

 return result

import time
import random
from requests.exceptions import RequestException

class FreddyAPIError(Exception):
 def __init__(self, message, status_code=None, error_type=None):
 self.message = message
 self.status_code = status_code
 self.error_type = error_type
 super().__init__(self.message)

def make_api_request_with_retry(endpoint, max_retries=3, backoff_factor=2):
 """Make API request with exponential backoff retry logic."""

 for attempt in range(max_retries + 1):
 try:
 response = requests.get(
 f"https://api.aitronos.com/v1/{endpoint}",
 headers={"X-API-Key": os.getenv('FREDDY_API_KEY')},
 timeout=30
 )

 # Handle different status codes
 if response.status_code == 200:
 return response.json()
 elif response.status_code == 401:
 raise FreddyAPIError("Invalid API key", 401, "AuthenticationError")
 elif response.status_code == 429:
 # Rate limited - wait and retry
 retry_after = int(response.headers.get('Retry-After', 60))
 if attempt < max_retries:
 time.sleep(retry_after)
 continue
 raise FreddyAPIError("Rate limit exceeded", 429, "RateLimitError")
 elif response.status_code >= 500:
 # Server error - retry with backoff
 if attempt < max_retries:
 wait_time = (backoff_factor ** attempt) + random.uniform(0, 1)
 time.sleep(wait_time)
 continue
 raise FreddyAPIError("Server error", response.status_code, "ServerError")
 else:
 # Other client errors - don't retry
 error_data = response.json() if response.content else {}
 raise FreddyAPIError(
 error_data.get('detail', 'Unknown error'),
 response.status_code,
 error_data.get('error_type', 'ClientError')
 )

 except RequestException as e:
 if attempt < max_retries:
 wait_time = (backoff_factor ** attempt) + random.uniform(0, 1)
 time.sleep(wait_time)
 continue
 raise FreddyAPIError(f"Network error: {str(e)}", None, "NetworkError")

 raise FreddyAPIError("Max retries exceeded", None, "RetryError")

import time
from datetime import datetime, timedelta

class RateLimitManager:
 def __init__(self, requests_per_minute=100):
 self.requests_per_minute = requests_per_minute
 self.requests = []

 def wait_if_needed(self):
 """Wait if we're approaching rate limits."""
 now = datetime.now()

 # Remove requests older than 1 minute
 self.requests = [req_time for req_time in self.requests
 if now - req_time < timedelta(minutes=1)]

 # If we're at the limit, wait
 if len(self.requests) >= self.requests_per_minute:
 oldest_request = min(self.requests)
 wait_time = 60 - (now - oldest_request).total_seconds()
 if wait_time > 0:
 time.sleep(wait_time)
 self.requests = [] # Reset after waiting

 # Record this request
 self.requests.append(now)

# Usage
rate_limiter = RateLimitManager(requests_per_minute=100)

def make_rate_limited_request(endpoint):
 rate_limiter.wait_if_needed()
 return make_api_request(endpoint)

def get_all_threads(freddy_client, limit=50):
 """Get all threads using pagination."""
 all_threads = []
 page = 1

 while True:
 response = freddy_client.get_threads(page=page, limit=limit)
 threads = response.get('data', [])

 if not threads:
 break

 all_threads.extend(threads)

 # Check if there are more pages
 pagination = response.get('pagination', {})
 if not pagination.get('has_next', False):
 break

 page += 1

 # Small delay between requests
 time.sleep(0.1)

 return all_threads

# Generator version for memory efficiency
def iter_all_threads(freddy_client, limit=50):
 """Generator that yields threads one by one."""
 page = 1

 while True:
 response = freddy_client.get_threads(page=page, limit=limit)
 threads = response.get('data', [])

 if not threads:
 break

 for thread in threads:
 yield thread

 pagination = response.get('pagination', {})
 if not pagination.get('has_next', False):
 break

 page += 1
 time.sleep(0.1)

# services/freddy_service.py
class FreddyService:
 def __init__(self, api_client, cache_client=None):
 self.api_client = api_client
 self.cache_client = cache_client

 def generate_ai_response(self, user_message, context=None):
 """Generate AI response with caching and error handling."""
 try:
 # Build the message payload
 messages = []
 if context:
 messages.append({"role": "system", "content": context})
 messages.append({"role": "user", "content": user_message})

 # Check cache first
 if self.cache_client:
 cache_key = self._build_cache_key(messages)
 cached_response = self.cache_client.get(cache_key)
 if cached_response:
 return json.loads(cached_response)

 # Make API request
 response = self.api_client.generate_response(
 model="gpt-4o",
 messages=messages,
 max_tokens=500,
 temperature=0.7
 )

 # Cache the response
 if self.cache_client:
 self.cache_client.setex(
 cache_key,
 3600, # 1 hour
 json.dumps(response)
 )

 return response

 except FreddyAPIError as e:
 # Log the error and handle gracefully
 logger.error(f"Freddy API error: {e.message}", extra={
 'status_code': e.status_code,
 'error_type': e.error_type
 })
 raise

 def _build_cache_key(self, messages):
 """Build a cache key from messages."""
 content = json.dumps(messages, sort_keys=True)
 return hashlib.md5(content.encode()).hexdigest()

import logging
import json
from datetime import datetime

# Configure structured logging
logging.basicConfig(
 level=logging.INFO,
 format='%(message)s'
)

logger = logging.getLogger(__name__)

class FreddyAPILogger:
 @staticmethod
 def log_api_request(endpoint, method, status_code, response_time, error=None):
 """Log API request with structured data."""
 log_data = {
 'timestamp': datetime.utcnow().isoformat(),
 'event': 'api_request',
 'endpoint': endpoint,
 'method': method,
 'status_code': status_code,
 'response_time_ms': response_time,
 'success': status_code < 400
 }

 if error:
 log_data['error'] = str(error)
 log_data['error_type'] = type(error).__name__

 logger.info(json.dumps(log_data))

 @staticmethod
 def log_usage_metrics(tokens_used, cost, model):
 """Log usage metrics for monitoring."""
 log_data = {
 'timestamp': datetime.utcnow().isoformat(),
 'event': 'usage_metrics',
 'tokens_used': tokens_used,
 'cost_usd': cost,
 'model': model
 }

 logger.info(json.dumps(log_data))

import unittest
from unittest.mock import Mock, patch
import responses

class TestFreddyService(unittest.TestCase):
 def setUp(self):
 self.mock_client = Mock()
 self.service = FreddyService(self.mock_client)

 @responses.activate
 def test_generate_response_success(self):
 """Test successful AI response generation."""
 # Mock the API response
 responses.add(
 responses.POST,
 "https://api.aitronos.com/v1/model/response",
 json={
 "status": "success",
 "data": {
 "message": {"role": "assistant", "content": "Hello!"},
 "usage": {"total_tokens": 10}
 }
 },
 status=200
 )

 result = self.service.generate_ai_response("Hello")

 self.assertEqual(result["data"]["message"]["content"], "Hello!")

 def test_error_handling(self):
 """Test error handling for API failures."""
 self.mock_client.generate_response.side_effect = FreddyAPIError(
 "Rate limit exceeded", 429, "RateLimitError"
 )

 with self.assertRaises(FreddyAPIError) as context:
 self.service.generate_ai_response("Hello")

 self.assertEqual(context.exception.status_code, 429)