import time from typing import Any import requests from requests import PreparedRequest, Response def record_session( url: str, method: str, headers: dict[str, Any] | None = None, body: str | dict[str, Any] | None = None, ) -> dict[str, Any]: """Record a single HTTP request/response session.""" start_time = time.time() session = requests.Session() req = PreparedRequest() req.prepare_method(method) req.prepare_url(url, None) req.prepare_headers(headers or {}) if body: if isinstance(body, dict): import json req.prepare_body(data=None, json=body) else: req.prepare_body(data=body, json=None) response: Response = session.send(request=req, timeout=30) end_time = time.time() latency_ms = int((end_time - start_time) * 1000) response_body = response.text try: response_body = response.json() except Exception: pass return { "request": { "method": method, "url": url, "headers": dict(req.headers), "body": body, "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(start_time)), }, "response": { "status_code": response.status_code, "headers": dict(response.headers), "body": response_body, "latency_ms": latency_ms, }, } def record_multiple( requests_config: list[dict[str, Any]], delay: float = 1.0, ) -> list[dict[str, Any]]: """Record multiple HTTP request/response sessions.""" results = [] for req_config in requests_config: url = req_config.get("url", "") method = req_config.get("method", "GET") headers = req_config.get("headers", {}) body = req_config.get("body") result = record_session( url=url, method=method, headers=headers, body=body, ) results.append(result) if delay > 0: time.sleep(delay) return results