app/core/timing.py

"""Timing utilities for performance monitoring."""
import time
import logging
from contextlib import contextmanager
from functools import wraps
from typing import Callable, Any

logger = logging.getLogger(__name__)


@contextmanager
def timer(operation_name: str, log_level: int = logging.INFO):
    """Context manager for timing operations.
    
    Usage:
        with timer("Database query"):
            result = db.query(...).all()
    
    Args:
        operation_name: Description of the operation being timed
        log_level: Logging level (default: INFO)
    """
    start = time.time()
    try:
        yield
    finally:
        elapsed = time.time() - start
        logger.log(log_level, f"{operation_name} took {elapsed:.3f}s")


def timed(operation_name: str = None):
    """Decorator to time function execution.
    
    Usage:
        @timed("Processing customer data")
        def process_customers(data):
            ...
    
    Args:
        operation_name: Description of the operation (defaults to function name)
    """
    def decorator(func: Callable) -> Callable:
        name = operation_name or f"{func.__module__}.{func.__name__}"
        
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            start = time.time()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                elapsed = time.time() - start
                if elapsed > 1.0:
                    logger.warning(f"{name} took {elapsed:.3f}s")
                elif elapsed > 0.5:
                    logger.info(f"{name} took {elapsed:.3f}s")
                else:
                    logger.debug(f"{name} took {elapsed:.3f}s")
        
        return wrapper
    return decorator


class QueryTimer:
    """Manual query timer for tracking specific query performance.
    
    Usage:
        qt = QueryTimer()
        qt.start("Customer list query")
        result = db.query(...).all()
        qt.stop()
    """
    
    def __init__(self):
        self.start_time = None
        self.operation_name = None
        self.logger = logging.getLogger("app.query.timing")
    
    def start(self, operation_name: str):
        """Start timing an operation."""
        self.operation_name = operation_name
        self.start_time = time.time()
        self.logger.debug(f"Started: {operation_name}")
    
    def stop(self) -> float:
        """Stop timing and log the result. Returns elapsed time."""
        if self.start_time is None:
            self.logger.warning("QueryTimer.stop() called without start()")
            return 0.0
        
        elapsed = time.time() - self.start_time
        
        if elapsed > 1.0:
            self.logger.warning(f"SLOW: {self.operation_name} took {elapsed:.3f}s")
        elif elapsed > 0.5:
            self.logger.info(f"{self.operation_name} took {elapsed:.3f}s")
        else:
            self.logger.debug(f"{self.operation_name} took {elapsed:.3f}s")
        
        self.start_time = None
        self.operation_name = None
        return elapsed