fastapi_endpoint_tracking_with_mongodb

FastAPI MongoDB Endpoint Tracking Implementation Guide

This rule provides a comprehensive guide for implementing endpoint tracking in FastAPI applications with MongoDB storage, analytics, and performance monitoring.

🏗️ Architecture Overview

The tracking system consists of:

1. Tracking Service - Core service that handles MongoDB operations
2. Endpoint Integration - Direct service calls in FastAPI endpoints
3. Analytics Endpoint - Query and aggregation interface
4. MongoDB Storage - Persistent storage with optimized indexes

📁 File Structure

backend/src/
├── services/
│   └── endpoint_tracking_service.py  # Core tracking service
├── api/v0/
│   ├── your_router.py                # Endpoints with tracking integration
└── utils/
    └── config.py                     # MongoDB configuration

🔧 Implementation Steps

Step 1: Configuration Setup

Add MongoDB tracking configuration to config.py:

class Settings(BaseSettings):
    # MongoDB Configuration (for endpoint tracking)
    MONGODB_CONNECTION_STRING: str = "your_mongodb_connection_string"
    MONGODB_DATABASE_NAME: str = "your_database_name"
    
    # Endpoint tracking configuration
    MONGODB_TRACKING_COLLECTION: str = "fastapi_tracking"
    MONGODB_MAX_POOL_SIZE: int = 100
    MONGODB_MIN_POOL_SIZE: int = 10
    MONGODB_MAX_IDLE_TIME_MS: int = 30000

Step 2: Create Tracking Service

Create endpoint_tracking_service.py:

import time
from datetime import datetime, timezone, timedelta
from typing import Dict, Any, Optional
from pymongo import MongoClient
from fastapi import Request, Response
import json

from utils.config import get_settings
from utils.logger import get_logger

logger = get_logger(__name__)

class EndpointTrackingService:
    """Service for tracking endpoint usage and storing analytics in MongoDB."""
    
    def __init__(self):
        self.settings = get_settings()
        self._init_database()
        
    def _init_database(self):
        """Initialize MongoDB connection for endpoint tracking"""
        logger.info("Initializing MongoDB connection for endpoint tracking...")
        
        try:
            self.mongo_client = MongoClient(
                self.settings.MONGODB_CONNECTION_STRING,
                maxPoolSize=self.settings.MONGODB_MAX_POOL_SIZE,
                minPoolSize=self.settings.MONGODB_MIN_POOL_SIZE,
                maxIdleTimeMS=self.settings.MONGODB_MAX_IDLE_TIME_MS
            )
            self.mongo_db = self.mongo_client[self.settings.MONGODB_DATABASE_NAME]
            self.tracking_collection = self.mongo_db[self.settings.MONGODB_TRACKING_COLLECTION]
            
            # Create indexes for better query performance
            self._create_indexes()
            
            logger.info("✅ MongoDB endpoint tracking connection established")
        except Exception as e:
            logger.error(f"❌ Failed to connect to MongoDB for endpoint tracking: {e}")
            raise
    
    def _create_indexes(self):
        """Create indexes for optimized querying"""
        try:
            # Check existing indexes first to avoid conflicts
            existing_indexes = list(self.tracking_collection.list_indexes())
            existing_index_names = [idx.get('name', '') for idx in existing_indexes]
            
            # Index on endpoint and timestamp for time-series queries
            if 'endpoint_1_timestamp_-1' not in existing_index_names:
                self.tracking_collection.create_index([
                    ("endpoint", 1),
                    ("timestamp", -1)
                ], name='endpoint_1_timestamp_-1')
            
            # Index on status_code for error tracking
            if 'status_code_1' not in existing_index_names:
                self.tracking_collection.create_index("status_code", name='status_code_1')
            
            # Index on response_time for performance monitoring
            if 'response_time_1' not in existing_index_names:
                self.tracking_collection.create_index("response_time", name='response_time_1')
            
            logger.info("✅ MongoDB indexes created/verified for endpoint tracking")
        except Exception as e:
            logger.warning(f"Failed to create indexes: {e}")

    async def track_request(
        self,
        request: Request,
        response: Response,
        endpoint_name: str,
        custom_data: Dict[str, Any],
        start_time: float
    ) -> None:
        """
        Generic method to track any endpoint request
        
        Args:
            request: FastAPI request object
            response: FastAPI response object
            endpoint_name: Name/path of the endpoint
            custom_data: Endpoint-specific data to track
            start_time: Request start timestamp
        """
        try:
            end_time = time.time()
            response_time = end_time - start_time
            
            # Base tracking data structure
            tracking_data = {
                # Basic request metadata
                "endpoint": endpoint_name,
                "method": request.method,
                "timestamp": datetime.now(timezone.utc),
                "response_time": response_time,
                "status_code": response.status_code,
                
                # Request details
                "user_agent": request.headers.get("user-agent"),
                "client_ip": self._get_client_ip(request),
                "query_params": dict(request.query_params),
                
                # Custom endpoint-specific data
                "custom_data": custom_data,
                
                # Environment info
                "environment": self.settings.ENVIRONMENT,
            }
            
            # Insert into MongoDB
            result = self.tracking_collection.insert_one(tracking_data)
            logger.info(f"📊 Request tracked for {endpoint_name}: {result.inserted_id}")
            
        except Exception as e:
            logger.error(f"Failed to track request for {endpoint_name}: {e}")
    
    def _get_client_ip(self, request: Request) -> str:
        """Extract client IP address from request headers"""
        # Check for forwarded IP headers (common in production)
        forwarded_for = request.headers.get("x-forwarded-for")
        if forwarded_for:
            return forwarded_for.split(",")[0].strip()
        
        real_ip = request.headers.get("x-real-ip")
        if real_ip:
            return real_ip
        
        # Fallback to direct client IP
        return getattr(request.client, "host", "unknown")

    async def get_endpoint_analytics(
        self,
        endpoint: Optional[str] = None,
        hours: int = 24
    ) -> Dict[str, Any]:
        """
        Get analytics for tracked endpoints
        
        Args:
            endpoint: Specific endpoint to analyze (optional)
            hours: Number of hours to look back
            
        Returns:
            Analytics data including request counts, response times, etc.
        """
        try:
            # Calculate time range
            end_time = datetime.now(timezone.utc)
            start_time = end_time - timedelta(hours=hours)
            
            # Build query
            query = {
                "timestamp": {
                    "$gte": start_time,
                    "$lte": end_time
                }
            }
            
            if endpoint:
                query["endpoint"] = endpoint
            
            # Get aggregated data
            pipeline = [
                {"$match": query},
                {"$group": {
                    "_id": "$endpoint",
                    "request_count": {"$sum": 1},
                    "avg_response_time": {"$avg": "$response_time"},
                    "max_response_time": {"$max": "$response_time"},
                    "min_response_time": {"$min": "$response_time"},
                    "error_count": {
                        "$sum": {
                            "$cond": [{"$gte": ["$status_code", 400]}, 1, 0]
                        }
                    }
                }}
            ]
            
            results = list(self.tracking_collection.aggregate(pipeline))
            
            return {
                "time_range": {
                    "start": start_time.isoformat(),
                    "end": end_time.isoformat(),
                    "hours": hours
                },
                "analytics": results
            }
            
        except Exception as e:
            logger.error(f"Failed to get endpoint analytics: {e}")
            return {"error": str(e)}

# Create singleton instance
endpoint_tracking_service = EndpointTrackingService()