strat-gameplay-webapp/backend/app/utils/CLAUDE.md

Utils - Shared Utilities and Helpers

Overview

Centralized utility functions and helpers used across the backend application. Provides logging configuration, authentication utilities, and shared helper functions.

Purpose

• Logging Setup: Centralized logging configuration with rotating file handlers
• Authentication: JWT token creation and verification
• Shared Helpers: Common utility functions used throughout the application

Module Structure

app/utils/
├── __init__.py          # Package marker
├── logging.py           # Logging configuration (rotating handlers)
└── auth.py              # JWT authentication utilities

Key Components

1. Logging Module (logging.py)

Configures application-wide logging with both console and file handlers.

Features

Rotating File Handlers:

• Daily log files: logs/app_YYYYMMDD.log
• Max file size: 10MB per file
• Keep last 5 backup files
• Format: YYYY-MM-DD HH:MM:SS - module.name - LEVEL - message

Dual Output:

• Console (INFO level): For development monitoring
• File (DEBUG level): For detailed debugging

Noisy Logger Silencing:

• SQLAlchemy engine: WARNING level only
• Socket.io: INFO level only
• Engine.io: WARNING level only

Usage

Application Startup (app/main.py):

from app.utils.logging import setup_logging

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown events"""
    logger.info("Starting Paper Dynasty Game Backend")
    setup_logging()  # Configure logging on startup
    await init_db()
    yield
    logger.info("Shutting down Paper Dynasty Game Backend")

Module-Level Loggers (Standard Pattern):

import logging

# For classes - include class name
logger = logging.getLogger(f'{__name__}.ClassName')

# For modules without classes - just module name
logger = logging.getLogger(f'{__name__}')

# Usage examples
logger.info("User connected")
logger.warning("Invalid request received")
logger.error("Failed to process action", exc_info=True)
logger.debug("Processing state transition")

Logging Pattern Examples Across Codebase:

# In game_engine.py
logger = logging.getLogger(f'{__name__}.GameEngine')

# In state_manager.py
logger = logging.getLogger(f'{__name__}.StateManager')

# In database operations
logger = logging.getLogger(f'{__name__}.DatabaseOperations')

# In WebSocket handlers
logger = logging.getLogger(f'{__name__}.handlers')

# In API routes
logger = logging.getLogger(f'{__name__}.games')

Configuration Details

Log Directory:

• Location: backend/logs/ (auto-created)
• Gitignored: Yes
• Naming: app_YYYYMMDD.log using Pendulum for UTC timestamps

Log Levels:

• Root logger: DEBUG (captures everything)
• Console handler: INFO (important messages only)
• File handler: DEBUG (full detail)
• SQLAlchemy engine: WARNING (reduce noise)
• Socket.io/Engine.io: INFO/WARNING (reduce noise)

Rotation:

• Size-based: 10MB max file size
• Time-based: Daily file names with UTC timestamps
• Backup count: Keep 5 most recent files
• Older logs automatically deleted

Log Format:

2025-10-31 14:23:45 - app.core.game_engine.GameEngine - INFO - Starting new game abc123
2025-10-31 14:23:46 - app.database.operations.DatabaseOperations - DEBUG - Executing query: INSERT INTO games...
2025-10-31 14:23:47 - app.websocket.handlers - ERROR - Connection failed: timeout

Implementation Details

def setup_logging() -> None:
    """Configure application logging"""

    # Create logs directory
    log_dir = "logs"
    os.makedirs(log_dir, exist_ok=True)

    # Log file name with date (Pendulum for UTC)
    now = pendulum.now('UTC')
    log_file = os.path.join(log_dir, f"app_{now.format('YYYYMMDD')}.log")

    # Formatter (consistent across handlers)
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    # Console handler (INFO for development)
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    # Rotating file handler (DEBUG for full detail)
    file_handler = logging.handlers.RotatingFileHandler(
        log_file,
        maxBytes=10 * 1024 * 1024,  # 10MB
        backupCount=5
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    # Configure root logger
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)
    root_logger.addHandler(console_handler)
    root_logger.addHandler(file_handler)

    # Silence noisy loggers
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
    logging.getLogger("socketio").setLevel(logging.INFO)
    logging.getLogger("engineio").setLevel(logging.WARNING)

2. Authentication Module (auth.py)

JWT token creation and verification for user authentication.

Features

Token Creation:

• JWT with HS256 algorithm
• 7-day expiration (configurable)
• Embeds arbitrary user data
• Uses Pendulum for UTC timestamps

Token Verification:

• Validates signature and expiration
• Returns decoded payload
• Raises JWTError on failure

Dependencies

from jose import jwt, JWTError  # Python-JOSE library
import pendulum                  # UTC timestamps
from app.config import get_settings  # Secret key from settings

Usage

Creating Tokens:

from app.utils.auth import create_token

# Create token with user data
user_data = {
    "user_id": 123,
    "username": "player1",
    "discord_id": "987654321",
    "league": "sba"
}
token = create_token(user_data)
# Returns: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Verifying Tokens:

from app.utils.auth import verify_token
from jose import JWTError

try:
    payload = verify_token(token)
    user_id = payload["user_id"]
    username = payload["username"]
except JWTError:
    # Token invalid or expired
    return {"error": "Invalid token"}

Integration with Discord OAuth (app/api/routes/auth.py):

from app.utils.auth import create_token

@router.post("/discord/callback")
async def discord_callback(code: str):
    # Exchange code for Discord token
    discord_token = await exchange_code_for_token(code)

    # Get Discord user info
    user_info = await get_discord_user(discord_token)

    # Create JWT for our application
    token = create_token({
        "user_id": user_info["id"],
        "username": user_info["username"],
        "discord_id": user_info["id"]
    })

    return {"token": token}

WebSocket Authentication (app/websocket/handlers.py):

from app.utils.auth import verify_token
from jose import JWTError

@sio.event
async def connect(sid, environ, auth):
    """Authenticate WebSocket connection"""
    try:
        # Verify token from auth dict
        token = auth.get("token")
        if not token:
            raise ConnectionRefusedError("Missing token")

        payload = verify_token(token)

        # Store user info in session
        await sio.save_session(sid, {
            "user_id": payload["user_id"],
            "username": payload["username"]
        })

        logger.info(f"User {payload['username']} connected")
    except JWTError:
        logger.warning("Invalid token on connection attempt")
        raise ConnectionRefusedError("Invalid token")

Implementation Details

Token Creation:

def create_token(user_data: Dict[str, Any]) -> str:
    """
    Create JWT token for user

    Args:
        user_data: User information to encode in token

    Returns:
        JWT token string
    """
    payload = {
        **user_data,
        "exp": pendulum.now('UTC').add(days=7).int_timestamp
    }
    token = jwt.encode(payload, settings.secret_key, algorithm="HS256")
    return token

Token Verification:

def verify_token(token: str) -> Dict[str, Any]:
    """
    Verify and decode JWT token

    Args:
        token: JWT token string

    Returns:
        Decoded token payload

    Raises:
        JWTError: If token is invalid or expired
    """
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
        return payload
    except JWTError as e:
        logger.warning(f"Invalid token: {e}")
        raise

Configuration Requirements (.env):

# Required for JWT signing
SECRET_KEY=your-secret-key-at-least-32-chars-long

Patterns & Conventions

Logging Pattern

Standard Pattern (Used Throughout Codebase):

import logging

# Module-level logger with class name
logger = logging.getLogger(f'{__name__}.ClassName')

class ClassName:
    def some_method(self):
        logger.info("Method called")
        logger.debug("Processing details...")
        logger.error("Something went wrong", exc_info=True)

Log Level Guidelines:

• DEBUG: Detailed diagnostic information (state changes, query details)
• INFO: General informational messages (connections, game events)
• WARNING: Warning messages (invalid input, deprecations)
• ERROR: Error messages (exceptions, failures)
• CRITICAL: Critical failures (system-level issues)

Exception Logging:

try:
    # operation
except Exception as e:
    logger.error(f"Operation failed: {e}", exc_info=True)
    raise

Authentication Pattern

Token in HTTP Headers:

# Client sends:
headers = {
    "Authorization": f"Bearer {token}"
}

Token in WebSocket Auth:

# Client connects with:
socket.io.connect("http://localhost:8000", {
    auth: {
        token: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
    }
})

FastAPI Dependency:

from fastapi import Depends, HTTPException, Header
from app.utils.auth import verify_token
from jose import JWTError

async def get_current_user(authorization: str = Header(None)):
    """Extract and verify user from Bearer token"""
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Missing or invalid authorization header")

    token = authorization.split(" ")[1]
    try:
        payload = verify_token(token)
        return payload
    except JWTError:
        raise HTTPException(401, "Invalid or expired token")

# Use in routes
@router.get("/protected")
async def protected_route(user = Depends(get_current_user)):
    return {"user_id": user["user_id"]}

Integration Points

Application Startup

main.py - Initialize logging on application startup:

from app.utils.logging import setup_logging

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Starting application")
    setup_logging()  # Must be called early
    await init_db()
    yield
    logger.info("Shutting down")

Module Initialization

Every Module with Logging:

import logging

logger = logging.getLogger(f'{__name__}.ClassName')

Modules Using Auth:

from app.utils.auth import create_token, verify_token
from jose import JWTError

WebSocket Handlers

Connection Authentication:

from app.utils.auth import verify_token

@sio.event
async def connect(sid, environ, auth):
    token = auth.get("token")
    payload = verify_token(token)  # Raises JWTError if invalid
    # Store session data

API Routes

Protected Endpoints:

from app.utils.auth import verify_token

@router.get("/games")
async def get_games(authorization: str = Header(None)):
    token = authorization.split(" ")[1]
    user = verify_token(token)
    # Use user data

Common Tasks

Adding a New Utility Module

1. Create Module File:

touch app/utils/new_utility.py

2. Implement Utility Functions:

# app/utils/new_utility.py
import logging

logger = logging.getLogger(f'{__name__}.NewUtility')

def utility_function(param: str) -> str:
    """
    Utility function description

    Args:
        param: Parameter description

    Returns:
        Return value description
    """
    logger.debug(f"Processing {param}")
    result = process(param)
    return result

3. Export from init.py (if needed for public API):

# app/utils/__init__.py
from .logging import setup_logging
from .auth import create_token, verify_token
from .new_utility import utility_function

__all__ = [
    "setup_logging",
    "create_token",
    "verify_token",
    "utility_function"
]

4. Add Tests:

# tests/unit/utils/test_new_utility.py
import pytest
from app.utils.new_utility import utility_function

def test_utility_function():
    result = utility_function("test")
    assert result == "expected"

Modifying Logging Configuration

Change Log Level:

# In setup_logging()
console_handler.setLevel(logging.DEBUG)  # Was INFO

Add Custom Handler:

# In setup_logging()
# Add email handler for critical errors
from logging.handlers import SMTPHandler

email_handler = SMTPHandler(
    mailhost=("smtp.example.com", 587),
    fromaddr="alerts@example.com",
    toaddrs=["admin@example.com"],
    subject="Critical Error in Game Backend"
)
email_handler.setLevel(logging.CRITICAL)
email_handler.setFormatter(formatter)
root_logger.addHandler(email_handler)

Change Rotation Policy:

# In setup_logging()
file_handler = logging.handlers.RotatingFileHandler(
    log_file,
    maxBytes=50 * 1024 * 1024,  # Change to 50MB
    backupCount=10              # Keep 10 backups
)

Changing Token Expiration

Modify create_token():

# In auth.py
def create_token(user_data: Dict[str, Any]) -> str:
    payload = {
        **user_data,
        "exp": pendulum.now('UTC').add(days=30).int_timestamp  # 30 days instead of 7
    }
    token = jwt.encode(payload, settings.secret_key, algorithm="HS256")
    return token

Adding Token Refresh

Create refresh_token() utility:

# In auth.py
def refresh_token(old_token: str) -> str:
    """
    Refresh an existing token with new expiration

    Args:
        old_token: Existing valid token

    Returns:
        New token with extended expiration

    Raises:
        JWTError: If old token is invalid
    """
    # Verify old token
    payload = verify_token(old_token)

    # Remove old expiration
    payload.pop("exp", None)

    # Create new token
    return create_token(payload)

Troubleshooting

Logging Issues

Problem: No logs appearing

• Check: setup_logging() called in main.py lifespan?
• Check: logs/ directory exists and is writable?
• Check: Using correct logger pattern: logger = logging.getLogger(f'{__name__}.ClassName')?

Problem: Too much SQLAlchemy noise in logs

• Solution: Already silenced in setup_logging() - check if being called
• Alternative: Increase SQLAlchemy logger level:

logging.getLogger("sqlalchemy.engine").setLevel(logging.ERROR)

Problem: Logs not rotating

• Check: File size exceeding 10MB?
• Check: RotatingFileHandler configured correctly?
• Debug: Check logs/ directory for backup files (app_YYYYMMDD.log.1, .2, etc.)

Problem: Daily log files not working

• Solution: Files are daily-named but don't auto-switch at midnight
• Enhancement: Use TimedRotatingFileHandler for true daily rotation:

file_handler = logging.handlers.TimedRotatingFileHandler(
    "logs/app.log",
    when="midnight",
    interval=1,
    backupCount=30  # Keep 30 days
)

Authentication Issues

Problem: JWTError: Invalid token

• Check: Token copied correctly (no whitespace)?
• Check: Token not expired (7-day limit)?
• Check: SECRET_KEY same on creation and verification?
• Check: Using correct algorithm (HS256)?

Problem: ImportError: cannot import name 'jwt' from 'jose'

• Solution: Install python-jose: pip install python-jose[cryptography]

Problem: Tokens working in Postman but not browser

• Check: CORS settings in config.py
• Check: Browser sending Authorization: Bearer <token> header?
• Check: Token not being stripped by middleware?

Problem: WebSocket auth failing

• Check: Token in auth object, not headers
• Client Code:

const socket = io("http://localhost:8000", {
    auth: {
        token: "YOUR_TOKEN_HERE"
    }
});

Pendulum Date Issues

Problem: AttributeError: 'DateTime' object has no attribute 'int_timestamp'

• Check: Using Pendulum 3.0+?
• Solution: int_timestamp is Pendulum 3.0 feature
• Alternative (Pendulum 2.x):

"exp": int(pendulum.now('UTC').add(days=7).timestamp())

Security Considerations

JWT Security

Secret Key Requirements:

• Minimum 32 characters
• Cryptographically random
• Never commit to git
• Rotate periodically

Generate Secure Key:

# Using Python
python -c "import secrets; print(secrets.token_urlsafe(32))"

# Using OpenSSL
openssl rand -base64 32

Token Best Practices:

• Always use HTTPS in production (tokens visible in HTTP)
• Short expiration times (7 days or less)
• Implement refresh token flow for long-lived sessions
• Never log token values
• Validate all claims on decode

Logging Security

Never Log:

• Passwords or secrets
• Complete JWT tokens
• Database credentials
• API keys
• User personal information

Safe Logging:

# ❌ DON'T
logger.info(f"User logged in with token: {token}")

# ✅ DO
logger.info(f"User {user_id} logged in")
logger.debug(f"Token prefix: {token[:10]}...")

Log Sanitization:

def sanitize_for_log(data: dict) -> dict:
    """Remove sensitive fields before logging"""
    sensitive_keys = ["password", "token", "secret", "api_key"]
    return {k: "***" if k in sensitive_keys else v for k, v in data.items()}

logger.info(f"Request data: {sanitize_for_log(request_data)}")

Testing

Unit Tests for Utilities

Test Structure:

tests/unit/utils/
├── test_logging.py      # Logging configuration tests
└── test_auth.py         # JWT creation/verification tests

Example: Testing Token Creation:

# tests/unit/utils/test_auth.py
import pytest
from app.utils.auth import create_token, verify_token
from jose import JWTError

def test_create_token():
    """Test JWT token creation"""
    user_data = {"user_id": 123, "username": "test"}
    token = create_token(user_data)

    assert token is not None
    assert isinstance(token, str)
    assert len(token) > 0

def test_verify_token_valid():
    """Test verifying valid token"""
    user_data = {"user_id": 123, "username": "test"}
    token = create_token(user_data)

    payload = verify_token(token)
    assert payload["user_id"] == 123
    assert payload["username"] == "test"
    assert "exp" in payload

def test_verify_token_invalid():
    """Test verifying invalid token"""
    with pytest.raises(JWTError):
        verify_token("invalid.token.here")

def test_token_expiration():
    """Test token expiration"""
    # Would need to mock time or wait 7 days
    # Use freezegun or similar for time-based tests
    pass

Example: Testing Logging Setup:

# tests/unit/utils/test_logging.py
import pytest
import logging
import os
from app.utils.logging import setup_logging

def test_setup_logging_creates_directory(tmp_path, monkeypatch):
    """Test that setup_logging creates logs directory"""
    monkeypatch.chdir(tmp_path)
    setup_logging()
    assert os.path.exists("logs")

def test_setup_logging_configures_handlers():
    """Test that handlers are configured"""
    setup_logging()
    root_logger = logging.getLogger()

    assert len(root_logger.handlers) >= 2
    assert any(isinstance(h, logging.StreamHandler) for h in root_logger.handlers)
    assert any(isinstance(h, logging.handlers.RotatingFileHandler) for h in root_logger.handlers)

def test_setup_logging_sets_levels():
    """Test that log levels are set correctly"""
    setup_logging()

    assert logging.getLogger("sqlalchemy.engine").level == logging.WARNING
    assert logging.getLogger("socketio").level == logging.INFO
    assert logging.getLogger("engineio").level == logging.WARNING

Examples

Complete Authentication Flow

# 1. User logs in via Discord OAuth
@router.post("/auth/discord/callback")
async def discord_callback(code: str):
    # Exchange code for Discord access token
    discord_token = await exchange_discord_code(code)

    # Get Discord user info
    user_info = await get_discord_user(discord_token)

    # Create our JWT token
    token = create_token({
        "user_id": user_info["id"],
        "username": user_info["username"],
        "discriminator": user_info["discriminator"]
    })

    return {"token": token}

# 2. Client stores token and uses for API calls
# Client code (JavaScript):
# localStorage.setItem("token", response.token)

# 3. Client connects to WebSocket with token
@sio.event
async def connect(sid, environ, auth):
    try:
        token = auth.get("token")
        payload = verify_token(token)

        await sio.save_session(sid, {
            "user_id": payload["user_id"],
            "username": payload["username"]
        })

        logger.info(f"User {payload['username']} connected")
    except JWTError:
        raise ConnectionRefusedError("Invalid token")

# 4. Client makes authenticated API requests
@router.get("/games/my-games")
async def get_my_games(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(401, "Missing authorization")

    token = authorization.replace("Bearer ", "")
    payload = verify_token(token)

    user_id = payload["user_id"]
    games = await db.get_games_for_user(user_id)
    return games

Custom Logger with Context

import logging
from contextvars import ContextVar

# Context variable for request ID
request_id_var: ContextVar[str] = ContextVar("request_id", default="")

class ContextFilter(logging.Filter):
    """Add request ID to log records"""
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

def setup_logging():
    """Enhanced logging with request context"""
    # Standard setup...
    formatter = logging.Formatter(
        '%(asctime)s - %(request_id)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    # Add context filter
    context_filter = ContextFilter()
    console_handler.addFilter(context_filter)
    file_handler.addFilter(context_filter)

    # ... rest of setup

# In middleware/handler
import uuid
from contextvars import ContextVar

@app.middleware("http")
async def request_id_middleware(request, call_next):
    request_id = str(uuid.uuid4())
    request_id_var.set(request_id)
    response = await call_next(request)
    return response

# Now all logs in that request context include request_id

Performance Considerations

Logging Performance

Avoid String Formatting Before Logging:

# ❌ DON'T - always formats even if not logged
logger.debug("Processing: " + expensive_operation())

# ✅ DO - only formats if debug enabled
logger.debug("Processing: %s", expensive_operation())

# ✅ ALSO GOOD - f-strings are evaluated lazily in some cases
logger.debug(f"Processing: {expensive_operation()}")  # But still evaluates

Use Lazy Evaluation:

if logger.isEnabledFor(logging.DEBUG):
    expensive_data = generate_debug_info()
    logger.debug(f"Debug info: {expensive_data}")

Token Verification Performance

Cache Decoded Tokens (for repeated verifications):

from functools import lru_cache

@lru_cache(maxsize=1000)
def verify_token_cached(token: str) -> Dict[str, Any]:
    """Cached token verification (use with caution)"""
    return verify_token(token)

# Clear cache periodically or on logout
verify_token_cached.cache_clear()

Note: Be careful with caching - can't invalidate specific tokens, and cache may return expired tokens if not cleared.

Related Documentation

• Main Backend CLAUDE.md: ../CLAUDE.md - Application architecture
• Config Module: ../config/CLAUDE.md - Settings and configuration
• WebSocket Module: ../websocket/CLAUDE.md - WebSocket integration
• API Routes: ../api/routes/CLAUDE.md - REST endpoint patterns

---

Key Principles:

1. Centralized Logging: All logging configuration in one place
2. Consistent Patterns: Use f'{__name__}.ClassName' everywhere
3. Security First: Never log sensitive data, use strong secret keys
4. Simple Utilities: Small, focused, reusable functions
5. Type Safety: Full type hints on all public functions

Testing: All utilities should have comprehensive unit tests covering normal and edge cases.