Как отслеживать время http запросов?

Отслеживание времени HTTP запросов

Отслеживание времени выполнения HTTP запросов критически важно для диагностики медленных запросов, выявления узких мест и оптимизации производительности приложения.

1. Встроенный модуль time

import time
import requests

def measure_http_request(url: str) -> dict:
    """Простое измерение времени запроса"""
    start_time = time.time()
    
    try:
        response = requests.get(url, timeout=10)
        total_time = time.time() - start_time
        
        return {
            "status_code": response.status_code,
            "total_time_ms": total_time * 1000,
            "response_size_bytes": len(response.content)
        }
    except requests.RequestException as e:
        total_time = time.time() - start_time
        return {
            "error": str(e),
            "total_time_ms": total_time * 1000
        }

result = measure_http_request("https://api.github.com")
print(f"Время: {result['total_time_ms']:.2f}ms")

2. requests библиотека встроенная статистика

import requests
from urllib.parse import urlparse

def track_request_timing(url: str):
    """Использование elapsed из requests"""
    response = requests.get(url)
    
    # elapsed — общее время (от запроса до ответа)
    print(f"Общее время: {response.elapsed.total_seconds():.3f}s")
    print(f"Статус: {response.status_code}")
    
    return response.elapsed.total_seconds()

track_request_timing("https://api.example.com/users")

3. Детальная диагностика с httpx

import httpx
import time

async def track_with_httpx(url: str):
    """httpx предоставляет детальную информацию о времени"""
    async with httpx.AsyncClient() as client:
        start = time.perf_counter()
        response = await client.get(url)
        elapsed = time.perf_counter() - start
        
        print(f"Время соединения: не напрямую доступно")
        print(f"Общее время: {elapsed * 1000:.2f}ms")
        print(f"Размер: {len(response.content)} bytes")

# Синхронный вариант
def track_with_httpx_sync(url: str):
    with httpx.Client() as client:
        start = time.perf_counter()
        response = client.get(url)
        elapsed = time.perf_counter() - start
        
        return {
            "elapsed_ms": elapsed * 1000,
            "status": response.status_code
        }

result = track_with_httpx_sync("https://api.github.com")
print(f"Ответ получен за {result['elapsed_ms']:.2f}ms")

4. perf_counter для точного измерения

import time
import requests
from typing import Callable, Any

def measure_performance(func: Callable) -> Callable:
    """Декоратор для измерения времени выполнения"""
    def wrapper(*args, **kwargs) -> tuple[Any, float]:
        # perf_counter лучше для относительных времён
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        
        return result, elapsed * 1000  # в миллисекундах
    return wrapper

@measure_performance
def fetch_user_data(user_id: int):
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return response.json()

data, elapsed_ms = fetch_user_data(123)
print(f"Запрос выполнен за {elapsed_ms:.2f}ms")

5. Логирование с временными метками

import logging
import time
import requests
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class HTTPLogger:
    def __init__(self):
        self.requests_log = []
    
    def log_request(self, method: str, url: str, duration_ms: float, status_code: int = None):
        """Логирует информацию о запросе"""
        timestamp = datetime.utcnow().isoformat()
        
        entry = {
            "timestamp": timestamp,
            "method": method,
            "url": url,
            "duration_ms": duration_ms,
            "status_code": status_code
        }
        
        self.requests_log.append(entry)
        
        if duration_ms > 1000:  # Если дольше 1 секунды
            logger.warning(f"Медленный запрос: {method} {url} - {duration_ms:.2f}ms")
        else:
            logger.info(f"{method} {url} - {duration_ms:.2f}ms")
    
    def get_slow_requests(self, threshold_ms: float = 500):
        """Возвращает запросы медленнее threshold"""
        return [r for r in self.requests_log if r["duration_ms"] > threshold_ms]

# Использование
http_logger = HTTPLogger()

start = time.perf_counter()
response = requests.get("https://api.github.com")
elapsed = (time.perf_counter() - start) * 1000

http_logger.log_request("GET", "https://api.github.com", elapsed, response.status_code)

6. Session с автоматическим логированием

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
import time

logger = logging.getLogger(__name__)

class LoggingHTTPAdapter(HTTPAdapter):
    """HTTP адаптер с логированием времени"""
    
    def send(self, request, *args, **kwargs):
        start = time.perf_counter()
        response = super().send(request, *args, **kwargs)
        elapsed = (time.perf_counter() - start) * 1000
        
        logger.info(
            f"{request.method} {request.url} - "
            f"Status: {response.status_code} - "
            f"Time: {elapsed:.2f}ms"
        )
        
        return response

def create_session_with_logging():
    """Создаёт сессию с логированием"""
    session = requests.Session()
    
    adapter = LoggingHTTPAdapter(
        max_retries=Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503, 504]
        )
    )
    
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

# Использование
session = create_session_with_logging()
response = session.get("https://api.github.com")

7. Мониторинг с прометеус

from prometheus_client import Histogram, Counter
import requests
import time

# Метрики
http_request_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint']
)

http_requests_total = Counter(
    'http_requests_total',
    'HTTP requests total',
    ['method', 'endpoint', 'status_code']
)

def fetch_with_metrics(method: str, url: str):
    """Запрос с метриками"""
    start = time.time()
    
    try:
        response = requests.request(method, url)
        duration = time.time() - start
        
        # Записываем метрики
        http_request_duration.labels(
            method=method,
            endpoint=url
        ).observe(duration)
        
        http_requests_total.labels(
            method=method,
            endpoint=url,
            status_code=response.status_code
        ).inc()
        
        return response
    except Exception as e:
        duration = time.time() - start
        http_request_duration.labels(
            method=method,
            endpoint=url
        ).observe(duration)
        raise

fetch_with_metrics("GET", "https://api.example.com/users")

8. Контекстный менеджер для измерения

from contextlib import contextmanager
import time
from typing import Generator

@contextmanager
def time_tracker(operation_name: str) -> Generator[dict, None, None]:
    """Контекстный менеджер для отслеживания времени"""
    timing_info = {"start": time.perf_counter()}
    
    try:
        yield timing_info
    finally:
        timing_info["end"] = time.perf_counter()
        timing_info["duration_ms"] = (timing_info["end"] - timing_info["start"]) * 1000
        
        print(f"{operation_name}: {timing_info['duration_ms']:.2f}ms")

# Использование
import requests

with time_tracker("API запрос"):
    response = requests.get("https://api.github.com")
    data = response.json()

9. Встроенный profiler

import cProfile
import pstats
import io
import requests

def profile_http_requests():
    """Профилирование всех операций"""
    pr = cProfile.Profile()
    pr.enable()
    
    # Делаем несколько запросов
    for i in range(5):
        requests.get(f"https://api.github.com/users/user{i}")
    
    pr.disable()
    
    # Выводим результаты
    s = io.StringIO()
    ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
    ps.print_stats(10)  # 10 самых медленных
    print(s.getvalue())

profile_http_requests()

Best Practices

• Используй perf_counter() для точного относительного времени
• Логируй медленные запросы (> 500ms) отдельно
• Мониторьте тренды — используй Prometheus, DataDog
• Кешируйте результаты если возможно
• Используй таймауты — указывай timeout на все запросы
• Пулируйте соединения — переиспользуйте Session

Отслеживание времени запросов — основа оптимизации и надёжности приложения.