Spaces:
Paused
Paused
File size: 4,400 Bytes
4a2ab42 d29a5a0 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """
Performance monitoring middleware
"""
import time
from functools import wraps
from typing import Any
from collections.abc import Callable
from core.logging import logger
def performance_monitor(func: Callable) -> Callable:
"""
Decorator to monitor function performance
"""
@wraps(func)
async def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
try:
result = await func(*args, **kwargs)
duration = time.time() - start_time
# Log slow operations
if duration > 1.0:
logger.warning(
f"Slow operation: {func.__name__} took {duration:.2f}s",
extra={
"operation": func.__name__,
"duration": duration,
"args_count": len(args),
"kwargs_count": len(kwargs),
},
)
elif duration > 0.1:
logger.info(f"Operation completed: {func.__name__} in {duration:.3f}s")
return result
except Exception as e:
duration = time.time() - start_time
logger.error(
f"Operation failed: {func.__name__} after {duration:.2f}s - {e}",
extra={
"operation": func.__name__,
"duration": duration,
"error": str(e),
},
)
raise
return wrapper
def database_query_monitor(func: Callable) -> Callable:
"""
Decorator to monitor database query performance
"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
# Log slow database operations
if duration > 0.5:
logger.warning(
f"Slow database operation: {func.__name__} took {duration:.3f}s",
extra={
"operation": func.__name__,
"duration": duration,
"query_type": "database",
},
)
return result
except Exception as e:
duration = time.time() - start_time
logger.error(
f"Database operation failed: {func.__name__} after {duration:.3f}s - {e}",
extra={
"operation": func.__name__,
"duration": duration,
"error": str(e),
"query_type": "database",
},
)
raise
return wrapper
class PerformanceTracker:
"""
Tracks application performance metrics
"""
def __init__(self):
self.metrics = {
"requests_total": 0,
"requests_duration_sum": 0.0,
"errors_total": 0,
"slow_requests": 0,
}
def record_request(self, duration: float, status_code: int):
"""Record HTTP request metrics"""
self.metrics["requests_total"] += 1
self.metrics["requests_duration_sum"] += duration
if duration > 5.0: # Slow request threshold
self.metrics["slow_requests"] += 1
if status_code >= 400:
self.metrics["errors_total"] += 1
def get_average_response_time(self) -> float:
"""Calculate average response time"""
if self.metrics["requests_total"] == 0:
return 0.0
return self.metrics["requests_duration_sum"] / self.metrics["requests_total"]
def get_error_rate(self) -> float:
"""Calculate error rate percentage"""
if self.metrics["requests_total"] == 0:
return 0.0
return (self.metrics["errors_total"] / self.metrics["requests_total"]) * 100
def get_metrics_summary(self) -> dict:
"""Get comprehensive metrics summary"""
return {
**self.metrics,
"average_response_time": self.get_average_response_time(),
"error_rate": self.get_error_rate(),
"slow_request_rate": (
self.metrics["slow_requests"] / max(self.metrics["requests_total"], 1)
)
* 100,
}
# Global performance tracker instance
performance_tracker = PerformanceTracker()
|