/

FastAPI Integration

FastAPI Integration

Integrate ByteHide Logs with FastAPI using middleware to track requests, user actions, and errors with minimal configuration.

Setup

Install the FastAPI integration:

Bash
pip install bytehide-logs[fastapi]

Add the ByteHide logging middleware to your FastAPI app:

Python
from fastapi import FastAPI
from bytehide_logs import Log, LogSettings
from bytehide_logs.fastapi import ByteHideLoggingMiddleware
from datetime import timedelta

app = FastAPI()

# Configure logging
settings = LogSettings(
    duplicate_suppression_window=timedelta(seconds=5),
    mask_sensitive_data=["password", "token", "api_key"]
)
Log.configure(settings)

# Add middleware
app.add_middleware(ByteHideLoggingMiddleware)

# Application metadata
Log.add_meta_context("service", "fastapi-app")
Log.add_meta_context("version", "1.0.0")

Middleware Configuration

The middleware automatically:

  • Generates and tracks correlation IDs
  • Logs incoming requests
  • Captures response times and status codes
  • Handles exceptions and errors
  • Cleans up context after request
Python
from fastapi import FastAPI
from bytehide_logs.fastapi import ByteHideLoggingMiddleware

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API with ByteHide Logs"
)

# Add ByteHide logging middleware
app.add_middleware(ByteHideLoggingMiddleware)

# Middleware configuration options
# - include_request_body: Include request body in logs (default: False for security)
# - include_response_body: Include response body in logs (default: False for security)
# - skip_paths: List of paths to exclude from logging (default: ["/health", "/docs"])

Request Tracking

Access the correlation ID in your endpoints:

Python
from fastapi import FastAPI, Request
from bytehide_logs import Log

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(item_id: int, request: Request):
    """Get item by ID with correlation tracking."""
    correlation_id = request.state.correlation_id
    
    Log.with_tags("item", "fetch") \
        .with_context("item_id", item_id) \
        .with_correlation_id(correlation_id) \
        .info("Fetching item")
    
    return {"item_id": item_id, "name": "Item Name"}

User Authentication

Track authenticated users with FastAPI security:

Python
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer
from bytehide_logs import Log, AuthUser

app = FastAPI()
security = HTTPBearer()

async def get_current_user(credentials = Depends(security)):
    """Get current authenticated user."""
    try:
        # Decode and verify token
        user_data = decode_token(credentials.credentials)
        
        # Identify user in logs
        user = AuthUser(
            id=user_data["user_id"],
            email=user_data["email"]
        )
        Log.identify(user)
        
        return user_data
    except Exception as e:
        Log.warning("Authentication failed", exception=e)
        raise HTTPException(status_code=401, detail="Not authenticated")

@app.get("/profile")
async def get_profile(user = Depends(get_current_user)):
    """Get user profile."""
    Log.with_tags("user", "profile").info("Fetching user profile")
    
    return {
        "user_id": user["user_id"],
        "email": user["email"]
    }

Exception Handling

Add custom exception handlers with logging:

Python
from fastapi import FastAPI, HTTPException
from bytehide_logs import Log

app = FastAPI()

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    """Handle HTTP exceptions with logging."""
    Log.with_tags("error", "http_exception") \
        .with_context("status_code", exc.status_code) \
        .with_context("detail", exc.detail) \
        .warning("HTTP exception occurred")
    
    return {
        "error": exc.detail,
        "status_code": exc.status_code,
        "correlation_id": request.state.get("correlation_id")
    }

@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
    """Handle unhandled exceptions with logging."""
    Log.with_tags("error", "unhandled") \
        .with_context("error_type", type(exc).__name__) \
        .error("Unhandled exception", exception=exc)
    
    return {
        "error": "Internal server error",
        "correlation_id": request.state.get("correlation_id")
    }, 500

Complete FastAPI Example

Python
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import HTTPBearer, HTTPAuthenticationCredentials
from bytehide_logs import Log, LogSettings, AuthUser
from datetime import timedelta
import uuid
from typing import Optional

# Initialize FastAPI app
app = FastAPI(
    title="User API",
    version="1.0.0"
)

# Configure logging
settings = LogSettings(
    duplicate_suppression_window=timedelta(seconds=5),
    mask_sensitive_data=["password", "token", "api_key"]
)
Log.configure(settings)

# Add middleware
from bytehide_logs.fastapi import ByteHideLoggingMiddleware
app.add_middleware(ByteHideLoggingMiddleware)

# Application metadata
Log.add_meta_context("service", "user-api")
Log.add_meta_context("environment", "production")

# Security
security = HTTPBearer()

# ============== Helper Functions ==============
async def get_current_user(
    credentials: Optional[HTTPAuthenticationCredentials] = Depends(security)
):
    """Get current authenticated user."""
    if not credentials:
        return None
    
    try:
        # Decode token (example)
        user_data = decode_jwt_token(credentials.credentials)
        
        # Identify user in logs
        user = AuthUser(
            id=user_data["user_id"],
            email=user_data["email"]
        )
        Log.identify(user)
        
        return user_data
    except Exception as e:
        Log.with_tags("auth", "decode_error").warning("Token decode failed", exception=e)
        raise HTTPException(status_code=401, detail="Not authenticated")

# ============== Exception Handlers ==============
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Handle HTTP exceptions."""
    Log.with_tags("error", "http") \
        .with_context("status_code", exc.status_code) \
        .with_context("detail", exc.detail) \
        .warning("HTTP exception")
    
    return {
        "error": exc.detail,
        "correlation_id": request.state.get("correlation_id")
    }

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Handle unhandled exceptions."""
    Log.with_tags("error", "unhandled") \
        .with_context("error_type", type(exc).__name__) \
        .error("Unhandled exception", exception=exc)
    
    return {
        "error": "Internal server error",
        "correlation_id": request.state.get("correlation_id")
    }, 500

# ============== Startup/Shutdown ==============
@app.on_event("startup")
async def startup_event():
    """Log application startup."""
    Log.info("FastAPI application starting")

@app.on_event("shutdown")
async def shutdown_event():
    """Log application shutdown."""
    Log.info("FastAPI application shutting down")

# ============== Auth Endpoints ==============
@app.post("/auth/login")
async def login(request: Request, email: str, password: str):
    """User login endpoint."""
    try:
        Log.with_tags("auth", "login") \
            .with_context("email", email) \
            .info("Login attempt")
        
        # Authenticate user (example)
        user_data = authenticate_user(email, password)
        
        # Generate token
        token = generate_jwt_token(user_data)
        
        # Identify user
        user = AuthUser(id=user_data["id"], email=user_data["email"])
        Log.identify(user)
        
        Log.with_tags("auth", "login", "success") \
            .with_context("user_id", user_data["id"]) \
            .info("Login successful")
        
        return {
            "token": token,
            "user_id": user_data["id"],
            "email": user_data["email"]
        }
        
    except Exception as e:
        Log.with_tags("auth", "login", "error") \
            .error("Login failed", exception=e)
        raise HTTPException(status_code=401, detail="Login failed")

@app.post("/auth/logout")
async def logout(request: Request):
    """User logout endpoint."""
    Log.with_tags("auth", "logout").info("Logout initiated")
    
    # Logout user
    Log.logout()
    
    Log.with_tags("auth", "logout", "success").info("Logout completed")
    
    return {"message": "Logged out"}

# ============== User Endpoints ==============
@app.get("/users/{user_id}")
async def get_user(user_id: str, request: Request):
    """Get user by ID."""
    correlation_id = request.state.correlation_id
    
    try:
        Log.with_tags("user", "fetch") \
            .with_context("user_id", user_id) \
            .with_correlation_id(correlation_id) \
            .info("Fetching user")
        
        user = fetch_user_from_db(user_id)
        
        Log.with_tags("user", "fetch", "success") \
            .with_context("user_id", user_id) \
            .with_correlation_id(correlation_id) \
            .info("User fetched")
        
        return user
        
    except Exception as e:
        Log.with_tags("user", "fetch", "error") \
            .with_context("user_id", user_id) \
            .with_correlation_id(correlation_id) \
            .error("Failed to fetch user", exception=e)
        raise HTTPException(status_code=404, detail="User not found")

@app.get("/profile")
async def get_profile(
    request: Request,
    user = Depends(get_current_user)
):
    """Get current user profile."""
    correlation_id = request.state.correlation_id
    
    Log.with_tags("user", "profile") \
        .with_context("user_id", user["user_id"]) \
        .with_correlation_id(correlation_id) \
        .info("Fetching profile")
    
    return {
        "user_id": user["user_id"],
        "email": user["email"],
        "name": user.get("name", "")
    }

@app.put("/profile")
async def update_profile(
    request: Request,
    user = Depends(get_current_user),
    name: Optional[str] = None,
    email: Optional[str] = None
):
    """Update current user profile."""
    correlation_id = request.state.correlation_id
    
    try:
        updates = {}
        if name:
            updates["name"] = name
        if email:
            updates["email"] = email
        
        Log.with_tags("user", "profile", "update") \
            .with_context("user_id", user["user_id"]) \
            .with_context("fields", list(updates.keys())) \
            .with_correlation_id(correlation_id) \
            .info("Updating profile")
        
        updated_user = update_user_in_db(user["user_id"], updates)
        
        Log.with_tags("user", "profile", "update", "success") \
            .with_context("user_id", user["user_id"]) \
            .with_correlation_id(correlation_id) \
            .info("Profile updated")
        
        return updated_user
        
    except Exception as e:
        Log.with_tags("user", "profile", "update", "error") \
            .with_context("user_id", user["user_id"]) \
            .with_correlation_id(correlation_id) \
            .error("Profile update failed", exception=e)
        raise HTTPException(status_code=500, detail="Update failed")

# ============== Order Endpoints ==============
@app.post("/orders")
async def create_order(
    request: Request,
    user = Depends(get_current_user),
    items: list = [],
    total: float = 0
):
    """Create new order."""
    correlation_id = request.state.correlation_id
    
    try:
        Log.with_tags("order", "create") \
            .with_context("user_id", user["user_id"]) \
            .with_context("item_count", len(items)) \
            .with_context("total", total) \
            .with_correlation_id(correlation_id) \
            .info("Creating order")
        
        # Validate order
        Log.with_tags("order", "validation") \
            .with_correlation_id(correlation_id) \
            .info("Validating order")
        validate_order({"items": items, "total": total})
        
        # Process payment
        Log.with_tags("order", "payment") \
            .with_context("total", total) \
            .with_correlation_id(correlation_id) \
            .info("Processing payment")
        payment = process_payment(total)
        
        # Create order
        order = create_order_in_db(
            user_id=user["user_id"],
            items=items,
            total=total,
            payment_id=payment["id"]
        )
        
        Log.with_tags("order", "create", "success") \
            .with_context("order_id", order["id"]) \
            .with_context("user_id", user["user_id"]) \
            .with_correlation_id(correlation_id) \
            .info("Order created")
        
        return {
            "order_id": order["id"],
            "status": "created",
            "correlation_id": correlation_id
        }
        
    except Exception as e:
        Log.with_tags("order", "create", "error") \
            .with_context("user_id", user["user_id"]) \
            .with_correlation_id(correlation_id) \
            .error("Order creation failed", exception=e)
        raise HTTPException(status_code=500, detail="Order creation failed")

# ============== Health Check ==============
@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy"}

# ============== Startup ==============
if __name__ == "__main__":
    import uvicorn
    Log.info("Starting FastAPI application")
    uvicorn.run(app, host="0.0.0.0", port=8000)

Best Practices

Always use correlation IDs for request tracking:

Python
@app.get("/api/endpoint")
async def endpoint(request: Request):
    correlation_id = request.state.correlation_id
    Log.with_correlation_id(correlation_id).info("Processing request")

Identify authenticated users:

Python
async def get_current_user(credentials = Depends(security)):
    user = decode_token(credentials.credentials)
    Log.identify(AuthUser(id=user["id"], email=user["email"]))
    return user

Use exception handlers for consistent error logging:

Python
@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception):
    Log.error("Unhandled exception", exception=exc)
    return error_response(str(exc))

Next Steps

Previous
Django