REST API Design

Caching Strategy

Multi-Level Caching

Client Request → CDN Cache → API Gateway Cache → Application Cache → Database
     │              │              │                    │              │
     ▼              ▼              ▼                    ▼              ▼
Static Assets   API Responses  Auth Tokens        Query Results   Raw Data
(Images, JS)    (1-5 minutes)  (15 minutes)      (30 seconds)    (Source)
TTL: 1 hour     TTL: Variable  TTL: Token Life   TTL: Variable   TTL: N/A
                        

Caching Implementation

# Multi-level caching system
import redis
from functools import wraps
import json
import hashlib
from typing import Any, Optional

class CacheManager:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.default_ttl = 300  # 5 minutes

    def generate_cache_key(self, prefix: str, *args, **kwargs) -> str:
        """Generate cache key from function arguments"""
        key_data = {
            'args': args,
            'kwargs': sorted(kwargs.items())
        }
        key_hash = hashlib.md5(json.dumps(key_data, sort_keys=True).encode()).hexdigest()
        return f"{prefix}:{key_hash}"

    async def get(self, key: str) -> Optional[Any]:
        """Get value from cache"""
        try:
            value = await self.redis.get(key)
            return json.loads(value) if value else None
        except Exception as e:
            logger.warning(f"Cache get error: {e}")
            return None

    async def set(self, key: str, value: Any, ttl: int = None) -> bool:
        """Set value in cache"""
        try:
            ttl = ttl or self.default_ttl
            serialized = json.dumps(value, default=str)
            return await self.redis.setex(key, ttl, serialized)
        except Exception as e:
            logger.warning(f"Cache set error: {e}")
            return False

    async def delete(self, pattern: str) -> int:
        """Delete keys matching pattern"""
        try:
            keys = await self.redis.keys(pattern)
            if keys:
                return await self.redis.delete(*keys)
            return 0
        except Exception as e:
            logger.warning(f"Cache delete error: {e}")
            return 0

    def cached(self, prefix: str, ttl: int = None,
               invalidate_on: list = None):
        """Decorator for caching function results"""
        def decorator(func):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                # Generate cache key
                cache_key = self.generate_cache_key(prefix, *args, **kwargs)

                # Try to get from cache
                cached_result = await self.get(cache_key)
                if cached_result is not None:
                    return cached_result

                # Execute function
                result = await func(*args, **kwargs)

                # Cache the result
                await self.set(cache_key, result, ttl)

                return result
            return wrapper
        return decorator

# Cache usage in API endpoints
cache_manager = CacheManager(redis_client)

@app.get("/api/v1/agents")
@cache_manager.cached("agents_list", ttl=60)  # 1 minute cache
async def list_agents(
    current_user: User = Depends(get_current_user),
    status: Optional[str] = None,
    limit: int = 50,
    offset: int = 0
):
    return await agent_service.list_agents(
        user=current_user,
        status=status,
        limit=limit,
        offset=offset
    )

@app.get("/api/v1/agents/{agent_id}")
@cache_manager.cached("agent_detail", ttl=30)  # 30 second cache
async def get_agent(
    agent_id: int,
    current_user: User = Depends(get_current_user)
):
    return await agent_service.get_agent(agent_id, current_user)

# Cache invalidation on updates
@app.put("/api/v1/agents/{agent_id}")
async def update_agent(
    agent_id: int,
    agent_data: AgentUpdate,
    current_user: User = Depends(get_current_user)
):
    # Update agent
    updated_agent = await agent_service.update_agent(
        agent_id, agent_data, current_user
    )

    # Invalidate related caches
    await cache_manager.delete(f"agent_detail:*{agent_id}*")
    await cache_manager.delete("agents_list:*")

    return updated_agent
                        

Database Query Optimization

Query Patterns

# Optimized database queries
from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import and_, or_, func

class OptimizedAgentService:
    def __init__(self, db_session):
        self.db = db_session

    async def list_agents_with_stats(self, user: User, filters: dict = None):
        """Optimized agent listing with statistics"""

        # Base query with necessary joins
        query = (
            select(Agent)
            .options(
                joinedload(Agent.group),  # Avoid N+1 queries for groups
                selectinload(Agent.packages).options(
                    # Only load package counts, not full objects
                    load_only(Package.id, Package.status)
                )
            )
            .where(Agent.deleted_at.is_(None))  # Soft delete filter
        )

        # Apply user-based permissions
        accessible_agent_ids = await self.get_accessible_agent_ids(user)
        if accessible_agent_ids:
            query = query.where(Agent.id.in_(accessible_agent_ids))

        # Apply filters efficiently
        if filters:
            if 'status' in filters:
                query = query.where(Agent.status == filters['status'])

            if 'platform' in filters:
                query = query.where(Agent.platform == filters['platform'])

            if 'search' in filters:
                search_term = f"%{filters['search']}%"
                query = query.where(
                    or_(
                        Agent.hostname.ilike(search_term),
                        Agent.ip_address.cast(String).like(search_term)
                    )
                )

        # Get total count efficiently (without loading all data)
        count_query = select(func.count(Agent.id)).select_from(
            query.subquery()
        )
        total_count = await self.db.scalar(count_query)

        # Apply pagination
        if 'limit' in filters:
            query = query.limit(filters['limit'])
        if 'offset' in filters:
            query = query.offset(filters['offset'])

        # Execute query
        result = await self.db.execute(query)
        agents = result.unique().scalars().all()

        # Calculate statistics in a single query
        stats = await self.get_agent_statistics(accessible_agent_ids)

        return {
            'agents': agents,
            'total': total_count,
            'statistics': stats
        }

    async def get_agent_statistics(self, agent_ids: List[int]):
        """Get agent statistics in a single optimized query"""

        stats_query = (
            select(
                Agent.status,
                func.count(Agent.id).label('count'),
                func.avg(Agent.cpu_usage).label('avg_cpu'),
                func.avg(Agent.memory_usage).label('avg_memory')
            )
            .where(Agent.id.in_(agent_ids))
            .group_by(Agent.status)
        )

        result = await self.db.execute(stats_query)
        stats_by_status = {row.status: {
            'count': row.count,
            'avg_cpu': row.avg_cpu,
            'avg_memory': row.avg_memory
        } for row in result}

        return {
            'by_status': stats_by_status,
            'total': sum(s['count'] for s in stats_by_status.values()),
            'online': stats_by_status.get('online', {}).get('count', 0)
        }

# Bulk operations for efficiency
class BulkOperations:
    def __init__(self, db_session):
        self.db = db_session

    async def bulk_update_agent_status(self, agent_updates: List[dict]):
        """Efficiently update multiple agents"""

        # Use bulk update for better performance
        stmt = (
            update(Agent)
            .where(Agent.id == bindparam('agent_id'))
            .values(
                status=bindparam('status'),
                last_seen=bindparam('last_seen'),
                updated_at=func.now()
            )
        )

        await self.db.execute(stmt, agent_updates)
        await self.db.commit()

    async def bulk_create_metrics(self, metrics: List[dict]):
        """Efficiently insert multiple metrics"""

        # Use bulk insert with ON CONFLICT handling
        stmt = insert(Metric).values(metrics)

        # Handle conflicts for duplicate timestamps
        stmt = stmt.on_conflict_do_update(
            index_elements=['agent_id', 'metric_name', 'timestamp'],
            set_={
                'value': stmt.excluded.value,
                'updated_at': func.now()
            }
        )

        await self.db.execute(stmt)
        await self.db.commit()
                        

Rate Limiting

Adaptive Rate Limiting

# Advanced rate limiting implementation
from typing import Dict, Optional
import time
import asyncio
from dataclasses import dataclass

@dataclass
class RateLimit:
    requests: int
    window_seconds: int
    burst_requests: int = None

class AdaptiveRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.default_limits = {
            'user': RateLimit(1000, 3600, 50),      # 1000/hour, burst 50
            'api_key': RateLimit(10000, 3600, 200), # 10000/hour, burst 200
            'ip': RateLimit(100, 3600, 10)          # 100/hour, burst 10
        }

    async def check_rate_limit(self, identifier: str,
                              limit_type: str = 'user') -> dict:
        """Check if request is within rate limits"""

        limit = self.default_limits.get(limit_type)
        if not limit:
            return {'allowed': True}

        now = time.time()
        window_start = int(now // limit.window_seconds) * limit.window_seconds

        # Keys for tracking
        window_key = f"rate_limit:{identifier}:{window_start}"
        burst_key = f"rate_limit_burst:{identifier}"

        # Use Redis pipeline for atomic operations
        async with self.redis.pipeline() as pipe:
            pipe.incr(window_key)
            pipe.expire(window_key, limit.window_seconds)
            pipe.incr(burst_key)
            pipe.expire(burst_key, 60)  # 1-minute burst window

            results = await pipe.execute()

        window_count = results[0]
        burst_count = results[2]

        # Check limits
        window_exceeded = window_count > limit.requests
        burst_exceeded = (limit.burst_requests and
                         burst_count > limit.burst_requests)

        if window_exceeded or burst_exceeded:
            reset_time = window_start + limit.window_seconds
            return {
                'allowed': False,
                'limit': limit.requests,
                'remaining': max(0, limit.requests - window_count),
                'reset_time': reset_time,
                'retry_after': max(1, reset_time - now)
            }

        return {
            'allowed': True,
            'limit': limit.requests,
            'remaining': limit.requests - window_count,
            'reset_time': window_start + limit.window_seconds
        }

# Rate limiting middleware
@app.middleware("http")
async def rate_limiting_middleware(request: Request, call_next):
    # Skip rate limiting for health checks
    if request.url.path in ['/health', '/metrics']:
        return await call_next(request)

    # Determine rate limit identifier
    identifier = None
    limit_type = 'ip'

    # Check for API key
    api_key = request.headers.get('X-API-Key')
    if api_key:
        identifier = f"api_key:{api_key}"
        limit_type = 'api_key'
    else:
        # Check for authenticated user
        auth_header = request.headers.get('Authorization')
        if auth_header and auth_header.startswith('Bearer '):
            try:
                token = auth_header.split(' ')[1]
                payload = jwt.decode(token, verify=False)  # Just for rate limiting
                identifier = f"user:{payload.get('user_id')}"
                limit_type = 'user'
            except:
                pass

    # Fall back to IP-based limiting
    if not identifier:
        # Get real IP (considering proxies)
        ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
        if not ip:
            ip = request.headers.get('X-Real-IP', '')
        if not ip:
            ip = request.client.host
        identifier = f"ip:{ip}"

    # Check rate limit
    rate_limiter = get_rate_limiter()
    result = await rate_limiter.check_rate_limit(identifier, limit_type)

    if not result['allowed']:
        return JSONResponse(
            status_code=429,
            content={
                "error": "Rate limit exceeded",
                "retry_after": result['retry_after']
            },
            headers={
                "X-RateLimit-Limit": str(result['limit']),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(int(result['reset_time'])),
                "Retry-After": str(int(result['retry_after']))
            }
        )

    # Continue with request
    response = await call_next(request)

    # Add rate limit headers to response
    if 'limit' in result:
        response.headers["X-RateLimit-Limit"] = str(result['limit'])
        response.headers["X-RateLimit-Remaining"] = str(result['remaining'])
        response.headers["X-RateLimit-Reset"] = str(int(result['reset_time']))

    return response