Add core modules: CLI, recorder, server, snapshot manager

api_snapshot/recorder/recorder.py

@@ -0,0 +1,85 @@
+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