Spaces:
Sleeping
Sleeping
File size: 4,793 Bytes
c385238 809aa75 c385238 809aa75 c385238 809aa75 c385238 809aa75 c385238 638b9bc c385238 638b9bc c385238 638b9bc c385238 638b9bc c385238 638b9bc c385238 638b9bc c385238 809aa75 c385238 809aa75 c385238 |
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 |
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import time
from collections import deque
import logging
class Metrics:
def __init__(self, total_requests=0, successful_requests=0, average_response_time=0.0, uptime=0.0):
"""
Initialize the Metrics class.
Args:
total_requests (int): Total number of requests.
successful_requests (int): Number of successful requests.
average_response_time (float): Average response time.
uptime (float): The total uptime in seconds.
"""
self.total_requests = total_requests
self.successful_requests = successful_requests
self.average_response_time = average_response_time
self.uptime = uptime
self.start_time = time.time()
def to_dict(self):
"""
Convert the metrics to a dictionary format.
"""
return {
"total_requests": self.total_requests,
"successful_requests": self.successful_requests,
"average_response_time": self.average_response_time,
"uptime": self.uptime + (time.time() - self.start_time)
}
@classmethod
def from_dict(cls, metrics_dict):
"""
Create a Metrics instance from a dictionary.
Args:
metrics_dict (dict): A dictionary containing the metrics.
"""
instance = cls(
total_requests=metrics_dict.get("total_requests", 0),
successful_requests=metrics_dict.get("successful_requests", 0),
average_response_time=metrics_dict.get("average_response_time", 0.0),
uptime=metrics_dict.get("uptime", 0.0)
)
return instance
class RateLimiter:
"""
Implements rate limiting using a sliding window approach.
Attributes:
requests_per_minute (int): Maximum allowed requests per minute
requests (deque): Queue storing request timestamps
Example:
>>> limiter = RateLimiter(requests_per_minute=60)
>>> limiter.can_proceed()
True
"""
def __init__(self, requests_per_minute: int = 60):
"""
Initialize rate limiter with requests per minute limit.
Args:
requests_per_minute (int): Maximum allowed requests per minute
"""
self.requests_per_minute = requests_per_minute
self.requests = deque()
def can_proceed(self) -> bool:
"""
Check if a new request can proceed based on rate limits.
Returns:
bool: True if request can proceed, False otherwise
"""
now = time.time()
# Remove requests older than 1 minute
while self.requests and self.requests[0] < now - 60:
self.requests.popleft()
if len(self.requests) < self.requests_per_minute:
self.requests.append(now)
return True
return False
class MetricsCollector:
"""
Collects and manages application metrics and rate limiting.
Combines Metrics and RateLimiter functionality to provide
comprehensive monitoring capabilities.
Example:
>>> collector = MetricsCollector()
>>> collector.record_request(success=True, response_time=0.1, tokens_used=100)
"""
def __init__(self):
"""Initialize metrics collector with default Metrics and RateLimiter."""
self.metrics = Metrics()
self.rate_limiter = RateLimiter()
def record_request(self, success: bool, response_time: float, tokens_used: int):
"""
Record a request and update metrics.
Args:
success (bool): Whether the request was successful
response_time (float): Request response time in seconds
tokens_used (int): Number of tokens used in the request
"""
self.metrics.total_requests += 1
if success:
self.metrics.successful_requests += 1
else:
# Ensure 'failed_requests' attribute is present in the Metrics class
if not hasattr(self.metrics, "failed_requests"):
self.metrics.failed_requests = 0
self.metrics.failed_requests += 1
# Update average response time
self.metrics.average_response_time = (
(self.metrics.average_response_time * (self.metrics.total_requests - 1) + response_time)
/ self.metrics.total_requests
)
# Ensure 'total_tokens_used' attribute is present in the Metrics class
if not hasattr(self.metrics, "total_tokens_used"):
self.metrics.total_tokens_used = 0
self.metrics.total_tokens_used += tokens_used |