Python API Development Best Practices: Building Robust and Scalable APIs
Building robust APIs is at the heart of modern software architecture. Whether you're developing microservices, integrating third-party systems, or creating data-driven applications, Python offers powerful frameworks that can handle enterprise-scale requirements. At Custom Logic, we've leveraged these patterns extensively in projects like the EOD Stock API, where performance and reliability are paramount.
Choosing the Right Framework: FastAPI vs Flask
The choice between FastAPI and Flask often determines your API's performance characteristics and development velocity. Each framework serves different use cases, and understanding their strengths helps you make informed architectural decisions.
FastAPI: Modern, Fast, and Type-Safe
FastAPI excels in scenarios requiring high performance, automatic documentation, and type safety. Here's a production-ready FastAPI implementation:
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import List, Optional
import asyncio
import httpx
app = FastAPI(
title="Stock Data API",
description="High-performance stock market data API",
version="1.0.0"
)
security = HTTPBearer()
class StockData(BaseModel):
symbol: str = Field(..., description="Stock symbol")
price: float = Field(..., gt=0, description="Current price")
volume: int = Field(..., ge=0, description="Trading volume")
timestamp: str = Field(..., description="Data timestamp")
class StockResponse(BaseModel):
data: List[StockData]
total: int
page: int
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
# Token validation logic here
if not credentials.credentials:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials"
)
return credentials.credentials
@app.get("/stocks/{symbol}", response_model=StockData)
async def get_stock_data(
symbol: str,
token: str = Depends(verify_token)
):
"""Retrieve real-time stock data for a specific symbol."""
async with httpx.AsyncClient() as client:
try:
# Simulate external API call
response = await client.get(f"https://api.example.com/stocks/{symbol}")
if response.status_code == 404:
raise HTTPException(
status_code=404,
detail=f"Stock symbol {symbol} not found"
)
return StockData(**response.json())
except httpx.RequestError:
raise HTTPException(
status_code=503,
detail="External service unavailable"
)
Flask: Flexible and Battle-Tested
Flask provides maximum flexibility for complex business logic and custom integrations. Here's a Flask implementation with similar functionality:
from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import wraps
import jwt
import requests
from datetime import datetime, timedelta
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["1000 per hour"]
)
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token:
return jsonify({'error': 'Token is missing'}), 401
try:
# Remove 'Bearer ' prefix
token = token.split(' ')[1] if token.startswith('Bearer ') else token
jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
except jwt.InvalidTokenError:
return jsonify({'error': 'Token is invalid'}), 401
return f(*args, **kwargs)
return decorated
@app.route('/stocks/<symbol>', methods=['GET'])
@token_required
@limiter.limit("100 per minute")
def get_stock_data(symbol):
"""Retrieve stock data with rate limiting and authentication."""
try:
# External API integration
response = requests.get(
f'https://api.example.com/stocks/{symbol}',
timeout=5
)
if response.status_code == 404:
return jsonify({'error': f'Stock symbol {symbol} not found'}), 404
response.raise_for_status()
return jsonify({
'symbol': symbol,
'data': response.json(),
'timestamp': datetime.utcnow().isoformat()
})
except requests.RequestException as e:
return jsonify({'error': 'External service error'}), 503
except Exception as e:
app.logger.error(f'Unexpected error: {str(e)}')
return jsonify({'error': 'Internal server error'}), 500
@app.errorhandler(429)
def ratelimit_handler(e):
return jsonify({'error': 'Rate limit exceeded', 'retry_after': e.retry_after}), 429
Authentication and Security Patterns
Security should be built into your API from the ground up. Here are essential patterns we implement across all Custom Logic projects:
JWT Authentication with Refresh Tokens
from datetime import datetime, timedelta
import jwt
from passlib.context import CryptContext
class AuthService:
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
self.access_token_expire = timedelta(minutes=15)
self.refresh_token_expire = timedelta(days=7)
def create_access_token(self, user_id: str) -> str:
expire = datetime.utcnow() + self.access_token_expire
payload = {
'user_id': user_id,
'exp': expire,
'type': 'access'
}
return jwt.encode(payload, self.secret_key, algorithm='HS256')
def create_refresh_token(self, user_id: str) -> str:
expire = datetime.utcnow() + self.refresh_token_expire
payload = {
'user_id': user_id,
'exp': expire,
'type': 'refresh'
}
return jwt.encode(payload, self.secret_key, algorithm='HS256')
def verify_token(self, token: str, token_type: str = 'access') -> dict:
try:
payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
if payload.get('type') != token_type:
raise jwt.InvalidTokenError('Invalid token type')
return payload
except jwt.ExpiredSignatureError:
raise jwt.InvalidTokenError('Token has expired')
except jwt.InvalidTokenError:
raise jwt.InvalidTokenError('Invalid token')
Input Validation and Sanitization
from pydantic import BaseModel, validator, Field
from typing import Optional
import re
class CreateUserRequest(BaseModel):
email: str = Field(..., description="User email address")
password: str = Field(..., min_length=8, description="User password")
name: str = Field(..., min_length=2, max_length=100, description="User full name")
phone: Optional[str] = Field(None, description="Phone number")
@validator('email')
def validate_email(cls, v):
email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}{{CONTENT}}#039;
if not re.match(email_pattern, v):
raise ValueError('Invalid email format')
return v.lower()
@validator('password')
def validate_password(cls, v):
if not re.search(r'[A-Z]', v):
raise ValueError('Password must contain at least one uppercase letter')
if not re.search(r'[a-z]', v):
raise ValueError('Password must contain at least one lowercase letter')
if not re.search(r'\d', v):
raise ValueError('Password must contain at least one digit')
return v
@validator('phone')
def validate_phone(cls, v):
if v is None:
return v
phone_pattern = r'^\+?1?[2-9]\d{2}[2-9]\d{2}\d{4}{{CONTENT}}#039;
if not re.match(phone_pattern, v):
raise ValueError('Invalid phone number format')
return v
Performance Optimization Strategies
Performance optimization becomes critical as your API scales. Here are proven patterns from our production systems:
Async Database Operations
import asyncpg
from typing import List, Optional
import asyncio
class AsyncDatabaseService:
def __init__(self, database_url: str):
self.database_url = database_url
self.pool: Optional[asyncpg.Pool] = None
async def initialize(self):
"""Initialize connection pool."""
self.pool = await asyncpg.create_pool(
self.database_url,
min_size=5,
max_size=20,
command_timeout=60
)
async def get_stock_data(self, symbols: List[str]) -> List[dict]:
"""Fetch multiple stock records efficiently."""
async with self.pool.acquire() as connection:
query = """
SELECT symbol, price, volume, last_updated
FROM stock_data
WHERE symbol = ANY($1::text[])
ORDER BY last_updated DESC
"""
rows = await connection.fetch(query, symbols)
return [dict(row) for row in rows]
async def batch_update_prices(self, price_updates: List[dict]):
"""Batch update stock prices for better performance."""
async with self.pool.acquire() as connection:
async with connection.transaction():
await connection.executemany(
"""
INSERT INTO stock_data (symbol, price, volume, last_updated)
VALUES ($1, $2, $3, $4)
ON CONFLICT (symbol)
DO UPDATE SET
price = EXCLUDED.price,
volume = EXCLUDED.volume,
last_updated = EXCLUDED.last_updated
""",
[(update['symbol'], update['price'],
update['volume'], update['timestamp'])
for update in price_updates]
)
Caching Strategies
import redis
import json
from typing import Any, Optional
import asyncio
class CacheService:
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url, decode_responses=True)
async def get_cached_data(self, key: str) -> Optional[dict]:
"""Retrieve cached data with JSON deserialization."""
try:
cached = self.redis.get(key)
return json.loads(cached) if cached else None
except (json.JSONDecodeError, redis.RedisError):
return None
async def cache_data(self, key: str, data: dict, ttl: int = 300):
"""Cache data with TTL (time to live)."""
try:
self.redis.setex(key, ttl, json.dumps(data))
except redis.RedisError:
# Log error but don't fail the request
pass
def cache_decorator(self, ttl: int = 300):
"""Decorator for automatic caching."""
def decorator(func):
async def wrapper(*args, **kwargs):
# Create cache key from function name and arguments
cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
# Try to get from cache first
cached_result = await self.get_cached_data(cache_key)
if cached_result:
return cached_result
# Execute function and cache result
result = await func(*args, **kwargs)
await self.cache_data(cache_key, result, ttl)
return result
return wrapper
return decorator
Error Handling and Monitoring
Robust error handling and monitoring are essential for production APIs. Here's how we implement comprehensive error management:
import logging
import traceback
from functools import wraps
from flask import jsonify, request
import time
# Configure structured logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class APIError(Exception):
"""Custom API exception with status code and message."""
def __init__(self, message: str, status_code: int = 500, payload=None):
super().__init__(message)
self.message = message
self.status_code = status_code
self.payload = payload
def handle_api_errors(f):
"""Decorator for consistent error handling across endpoints."""
@wraps(f)
def decorated_function(*args, **kwargs):
start_time = time.time()
try:
result = f(*args, **kwargs)
# Log successful requests
duration = time.time() - start_time
logger.info(f"Request completed: {request.method} {request.path} - "
f"Duration: {duration:.3f}s")
return result
except APIError as e:
logger.warning(f"API Error: {e.message} - Status: {e.status_code}")
response = {'error': e.message}
if e.payload:
response.update(e.payload)
return jsonify(response), e.status_code
except Exception as e:
# Log unexpected errors with full traceback
logger.error(f"Unexpected error in {f.__name__}: {str(e)}\n"
f"Traceback: {traceback.format_exc()}")
return jsonify({'error': 'Internal server error'}), 500
return decorated_function
# Usage example
@app.route('/api/stocks/<symbol>')
@handle_api_errors
def get_stock_info(symbol):
if not symbol.isalpha():
raise APIError('Invalid symbol format', 400)
# Your business logic here
return jsonify({'symbol': symbol, 'price': 150.00})
Testing Strategies
Comprehensive testing ensures your API remains reliable as it evolves. Here's our testing approach:
import pytest
import asyncio
from httpx import AsyncClient
from unittest.mock import AsyncMock, patch
class TestStockAPI:
@pytest.fixture
async def client(self):
"""Create test client."""
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
@pytest.fixture
def mock_auth_token(self):
"""Mock authentication token."""
return "Bearer valid-test-token"
async def test_get_stock_data_success(self, client, mock_auth_token):
"""Test successful stock data retrieval."""
with patch('httpx.AsyncClient.get') as mock_get:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = {
'symbol': 'AAPL',
'price': 150.00,
'volume': 1000000,
'timestamp': '2025-01-17T10:00:00Z'
}
mock_get.return_value = mock_response
response = await client.get(
"/stocks/AAPL",
headers={"Authorization": mock_auth_token}
)
assert response.status_code == 200
data = response.json()
assert data['symbol'] == 'AAPL'
assert data['price'] == 150.00
async def test_get_stock_data_not_found(self, client, mock_auth_token):
"""Test handling of non-existent stock symbol."""
with patch('httpx.AsyncClient.get') as mock_get:
mock_response = AsyncMock()
mock_response.status_code = 404
mock_get.return_value = mock_response
response = await client.get(
"/stocks/INVALID",
headers={"Authorization": mock_auth_token}
)
assert response.status_code == 404
assert 'not found' in response.json()['detail'].lower()
async def test_unauthorized_access(self, client):
"""Test API security with invalid token."""
response = await client.get("/stocks/AAPL")
assert response.status_code == 401
Production Deployment Considerations
When deploying Python APIs to production, several factors ensure reliability and performance:
Configuration Management
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
"""Application settings with environment variable support."""
database_url: str
redis_url: str
secret_key: str
debug: bool = False
log_level: str = "INFO"
api_rate_limit: str = "1000/hour"
class Config:
env_file = ".env"
case_sensitive = False
settings = Settings()
Health Checks and Monitoring
@app.get("/health")
async def health_check():
"""Comprehensive health check endpoint."""
checks = {
'database': await check_database_connection(),
'redis': await check_redis_connection(),
'external_apis': await check_external_services()
}
all_healthy = all(checks.values())
status_code = 200 if all_healthy else 503
return JSONResponse(
content={
'status': 'healthy' if all_healthy else 'unhealthy',
'checks': checks,
'timestamp': datetime.utcnow().isoformat()
},
status_code=status_code
)
Conclusion
Building production-ready Python APIs requires careful attention to architecture, security, performance, and maintainability. The patterns demonstrated here have proven effective in real-world applications, from high-frequency trading systems to enterprise business applications.
At Custom Logic, we apply these best practices across all our API development projects. Whether you're building a new API from scratch or optimizing an existing system, these patterns provide a solid foundation for scalable, secure, and maintainable solutions.
The key to successful API development lies in choosing the right tools for your specific requirements, implementing robust security from the start, and maintaining comprehensive testing coverage. By following these practices, you'll build APIs that can handle enterprise-scale demands while remaining maintainable and secure.