HTTP REST Client with Retry Logic (python, written by Claude Code)

Claude Code wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: HTTP REST Client with Retry Logic

Write a program that sends HTTP requests to REST APIs with configurable retry logic, response validation, request chaining, and comprehensive logging of request/response cycles.

FUNCTIONAL REQUIREMENTS:
- Accept a URL as a command-line argument with --method flag (GET, POST, PUT, PATCH, DELETE; default: GET)
- Support sending request body via --data flag (JSON string) or --file flag (read body from a file)
- Support custom headers via --header flag (repeatable, e.g., --header "Content-Type: application/json" --header "Authorization: Bearer token")
- Implement configurable retry logic: --retries flag (default 3), --backoff flag for backoff strategy (linear, exponential; default: exponential), --retry-delay flag for initial delay in seconds (default 1)
- Retry only on configurable HTTP status codes via --retry-on flag (default: 429, 500, 502, 503, 504)
- Support request timeout via --timeout flag (default 30 seconds)
- Support request chaining from a JSON recipe file via --chain flag: execute a sequence of requests where later requests can reference values from earlier responses using JSONPath expressions
- Log complete request/response cycles: method, URL, headers, body, status code, response headers, response body, timing, and retry attempts
- Support response validation via --expect-status flag (e.g., --expect-status 200) and --expect-json flag (JSONPath expression that must exist in the response)
- Print formatted response to console: status code, headers, and pretty-printed body (auto-detect JSON and format it)
- Save the full request/response log as JSON with --output flag (default: request_log.json)
- If no URL is given, demonstrate the client by sending requests to a public test API (httpbin.org or similar), showing GET, POST with JSON body, retry on simulated failure, and a simple request chain
- Handle errors: connection refused, DNS failure, SSL errors, malformed JSON responses, and redirect loops

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

"""
HTTP REST Client with Retry Logic

A robust HTTP client that supports:
- GET, POST, PUT, PATCH, DELETE requests with retry/backoff
- Request chaining (use response data in subsequent requests)
- Response validation with custom rules
- Structured logging of all requests/responses
"""

import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Union

import requests
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
    before_sleep_log,
    RetryError,
)

# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------

LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s - %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
logger = logging.getLogger("rest_client")


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------

@dataclass
class ValidationRule:
    """A single validation rule applied to an HTTP response."""
    name: str
    check: Callable[[requests.Response], bool]
    message: str = ""


class ValidationError(Exception):
    """Raised when a response fails validation."""

    def __init__(self, failed_rules: List[str]):
        self.failed_rules = failed_rules
        super().__init__(f"Validation failed: {', '.join(failed_rules)}")


# ---------------------------------------------------------------------------
# Built-in validators
# ---------------------------------------------------------------------------

def status_ok() -> ValidationRule:
    """Validate that the status code is 2xx."""
    return ValidationRule(
        name="status_ok",
        check=lambda r: 200 <= r.status_code < 300,
        message="Expected 2xx status code",
    )


def status_equals(code: int) -> ValidationRule:
    """Validate that the status code matches exactly."""
    return ValidationRule(
        name=f"status_{code}",
        check=lambda r: r.status_code == code,
        message=f"Expected status code {code}",
    )


def json_has_key(key: str) -> ValidationRule:
    """Validate that the JSON response contains a specific key."""
    def _check(r: requests.Response) -> bool:
        try:
            return key in r.json()
        except (ValueError, KeyError):
            return False

    return ValidationRule(
        name=f"has_key_{key}",
        check=_check,
        message=f"Expected JSON key '{key}' in response",
    )


def content_type_is(expected: str) -> ValidationRule:
    """Validate the Content-Type header."""
    return ValidationRule(
        name="content_type",
        check=lambda r: expected in r.headers.get("Content-Type", ""),
        message=f"Expected Content-Type containing '{expected}'",
    )


def max_response_time(seconds: float) -> ValidationRule:
    """Validate that the response arrived within a time limit."""
    return ValidationRule(
        name="max_response_time",
        check=lambda r: r.elapsed.total_seconds() <= seconds,
        message=f"Response took longer than {seconds}s",
    )


# ---------------------------------------------------------------------------
# Request / Response wrappers
# ---------------------------------------------------------------------------

@dataclass
class RequestConfig:
    """Configuration for a single HTTP request."""
    method: str
    url: str
    headers: Optional[Dict[str, str]] = None
    params: Optional[Dict[str, str]] = None
    json_body: Optional[Any] = None
    data: Optional[Any] = None
    timeout: float = 30.0
    validators: List[ValidationRule] = field(default_factory=list)
    label: str = ""


@dataclass
class ResponseRecord:
    """Stores details about a completed request/response cycle."""
    label: str
    method: str
    url: str
    status_code: int
    elapsed_seconds: float
    headers: Dict[str, str]
    body: Any
    ok: bool
    validation_errors: List[str] = field(default_factory=list)


# ---------------------------------------------------------------------------
# REST Client
# ---------------------------------------------------------------------------

class RestClient:
    """HTTP REST client with retry logic, chaining, and validation."""

    def __init__(
        self,
        base_url: str = "",
        default_headers: Optional[Dict[str, str]] = None,
        max_retries: int = 3,
        backoff_multiplier: float = 1.0,
        backoff_max: float = 30.0,
        retry_status_codes: Optional[List[int]] = None,
    ):
        self.base_url = base_url.rstrip("/")
        self.default_headers = default_headers or {}
        self.max_retries = max_retries
        self.backoff_multiplier = backoff_multiplier
        self.backoff_max = backoff_max
        self.retry_status_codes = retry_status_codes or [429, 500, 502, 503, 504]
        self.session = requests.Session()
        self.session.headers.update(self.default_headers)
        self.history: List[ResponseRecord] = []

    # ---- internal ---------------------------------------------------------

    def _build_url(self, path: str) -> str:
        if path.startswith("http://") or path.startswith("https://"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _should_retry(self, response: requests.Response) -> bool:
        return response.status_code in self.retry_status_codes

    def _validate(
        self, response: requests.Response, rules: List[ValidationRule]
    ) -> List[str]:
        errors: List[str] = []
        for rule in rules:
            try:
                if not rule.check(response):
                    msg = rule.message or rule.name
                    errors.append(msg)
                    logger.warning("Validation failed [%s]: %s", rule.name, msg)
            except Exception as exc:
                errors.append(f"{rule.name}: {exc}")
        return errors

    def _log_request(self, cfg: RequestConfig) -> None:
        logger.info(
            ">>> %s %s | label=%s",
            cfg.method.upper(),
            self._build_url(cfg.url),
            cfg.label or "(none)",
        )
        if cfg.json_body:
            logger.debug("    body: %s", json.dumps(cfg.json_body, indent=2))

    def _log_response(self, record: ResponseRecord) -> None:
        level = logging.INFO if record.ok else logging.WARNING
        logger.log(
            level,
            "<<< %d %s (%.3fs) | label=%s | validation_errors=%s",
            record.status_code,
            record.url,
            record.elapsed_seconds,
            record.label or "(none)",
            record.validation_errors or "none",
        )

    # ---- core request -----------------------------------------------------

    def request(self, cfg: RequestConfig) -> ResponseRecord:
        """
        Execute a single HTTP request with retry/backoff and validation.
        """
        self._log_request(cfg)
        url = self._build_url(cfg.url)
        merged_headers = {**self.default_headers, **(cfg.headers or {})}

        @retry(
            stop=stop_after_attempt(self.max_retries),
            wait=wait_exponential(
                multiplier=self.backoff_multiplier, max=self.backoff_max
            ),
            retry=retry_if_exception_type(requests.exceptions.ConnectionError),
            before_sleep=before_sleep_log(logger, logging.WARNING),
            reraise=True,
        )
        def _do_request() -> requests.Response:
            resp = self.session.request(
                method=cfg.method.upper(),
                url=url,
                headers=merged_headers,
                params=cfg.params,
                json=cfg.json_body,
                data=cfg.data,
                timeout=cfg.timeout,
            )
            if self._should_retry(resp):
                logger.warning(
                    "Retryable status %d from %s – will retry", resp.status_code, url
                )
                raise requests.exceptions.ConnectionError(
                    f"Retryable status {resp.status_code}"
                )
            return resp

        try:
            response = _do_request()
        except RetryError as exc:
            logger.error("All retries exhausted for %s %s", cfg.method.upper(), url)
            raise

        # Parse body
        try:
            body = response.json()
        except ValueError:
            body = response.text

        validation_errors = self._validate(response, cfg.validators)
        record = ResponseRecord(
            label=cfg.label,
            method=cfg.method.upper(),
            url=url,
            status_code=response.status_code,
            elapsed_seconds=response.elapsed.total_seconds(),
            headers=dict(response.headers),
            body=body,
            ok=response.ok and len(validation_errors) == 0,
            validation_errors=validation_errors,
        )
        self._log_response(record)
        self.history.append(record)

        if validation_errors:
            raise ValidationError(validation_errors)

        return record

    # ---- convenience methods ----------------------------------------------

    def get(self, url: str, **kwargs) -> ResponseRecord:
        return self.request(RequestConfig(method="GET", url=url, **kwargs))

    def post(self, url: str, **kwargs) -> ResponseRecord:
        return self.request(RequestConfig(method="POST", url=url, **kwargs))

    def put(self, url: str, **kwargs) -> ResponseRecord:
        return self.request(RequestConfig(method="PUT", url=url, **kwargs))

    def patch(self, url: str, **kwargs) -> ResponseRecord:
        return self.request(RequestConfig(method="PATCH", url=url, **kwargs))

    def delete(self, url: str, **kwargs) -> ResponseRecord:
        return self.request(RequestConfig(method="DELETE", url=url, **kwargs))

    # ---- request chaining -------------------------------------------------

    def chain(
        self,
        configs: List[RequestConfig],
        context: Optional[Dict[str, Any]] = None,
        extractor: Optional[Callable[[ResponseRecord, Dict[str, Any]], Dict[str, Any]]] = None,
    ) -> List[ResponseRecord]:
        """
        Execute a sequence of requests, passing context between them.

        *extractor* receives the latest ResponseRecord and the current
        context dict, and must return the updated context dict.  Subsequent
        RequestConfig objects may reference context values via ``{key}``
        placeholders in their URL, headers, or JSON body.
        """
        ctx = dict(context or {})
        results: List[ResponseRecord] = []

        for idx, cfg in enumerate(configs):
            # Interpolate context into URL
            cfg.url = cfg.url.format(**ctx) if ctx else cfg.url

            # Interpolate context into headers
            if cfg.headers:
                cfg.headers = {
                    k: v.format(**ctx) if isinstance(v, str) else v
                    for k, v in cfg.headers.items()
                }

            # Interpolate context into JSON body (string values only)
            if cfg.json_body and isinstance(cfg.json_body, dict):
                cfg.json_body = {
                    k: v.format(**ctx) if isinstance(v, str) else v
                    for k, v in cfg.json_body.items()
                }

            logger.info("Chain step %d/%d: %s", idx + 1, len(configs), cfg.label or cfg.url)
            record = self.request(cfg)
            results.append(record)

            if extractor:
                ctx = extractor(record, ctx)

        return results

    # ---- history ----------------------------------------------------------

    def print_history(self) -> None:
        """Pretty-print the request history."""
        print(f"\n{'='*70}")
        print(f" Request History ({len(self.history)} requests)")
        print(f"{'='*70}")
        for i, rec in enumerate(self.history, 1):
            status_indicator = "OK" if rec.ok else "FAIL"
            print(
                f"  [{i}] {status_indicator} | {rec.method} {rec.url} "
                f"-> {rec.status_code} ({rec.elapsed_seconds:.3f}s) "
                f"| {rec.label}"
            )
            if rec.validation_errors:
                for err in rec.validation_errors:
                    print(f"       ! {err}")
        print(f"{'='*70}\n")


# ---------------------------------------------------------------------------
# Demo / main
# ---------------------------------------------------------------------------

def main():
    """Demonstrate the REST client features."""
    client = RestClient(
        base_url="https://jsonplaceholder.typicode.com",
        default_headers={"Accept": "application/json"},
        max_retries=3,
        backoff_multiplier=0.5,
    )

    # 1. Simple GET with validation
    print("\n--- Simple GET with validation ---")
    try:
        record = client.get(
            "/posts/1",
            label="get_post_1",
            validators=[status_ok(), json_has_key("title"), content_type_is("json")],
        )
        print(f"Post title: {record.body.get('title', 'N/A')}")
    except (ValidationError, Exception) as exc:
        print(f"Request failed: {exc}")

    # 2. POST with body
    print("\n--- POST with body ---")
    try:
        record = client.post(
            "/posts",
            label="create_post",
            json_body={"title": "Hello", "body": "World", "userId": 1},
            validators=[status_equals(201)],
        )
        print(f"Created post ID: {record.body.get('id', 'N/A')}")
    except (ValidationError, Exception) as exc:
        print(f"Request failed: {exc}")

    # 3. Request chaining — get a user, then get their posts
    print("\n--- Request chaining ---")

    chain_configs = [
        RequestConfig(
            method="GET",
            url="/users/1",
            label="get_user",
            validators=[status_ok()],
        ),
        RequestConfig(
            method="GET",
            url="/users/{user_id}/posts",
            label="get_user_posts",
            validators=[status_ok()],
        ),
    ]

    def extract_user_id(record: ResponseRecord, ctx: Dict[str, Any]) -> Dict[str, Any]:
        if isinstance(record.body, dict) and "id" in record.body:
            ctx["user_id"] = record.body["id"]
        return ctx

    try:
        results = client.chain(chain_configs, extractor=extract_user_id)
        if len(results) >= 2 and isinstance(results[1].body, list):
            print(f"User has {len(results[1].body)} posts")
    except Exception as exc:
        print(f"Chain failed: {exc}")

    # 4. PUT request
    print("\n--- PUT request ---")
    try:
        record = client.put(
            "/posts/1",
            label="update_post",
            json_body={"id": 1, "title": "Updated", "body": "Content", "userId": 1},
            validators=[status_ok()],
        )
        print(f"Updated post title: {record.body.get('title', 'N/A')}")
    except (ValidationError, Exception) as exc:
        print(f"Request failed: {exc}")

    # 5. DELETE request
    print("\n--- DELETE request ---")
    try:
        record = client.delete(
            "/posts/1",
            label="delete_post",
            validators=[status_ok()],
        )
        print(f"Delete status: {record.status_code}")
    except (ValidationError, Exception) as exc:
        print(f"Request failed: {exc}")

    # Print full history
    client.print_history()


if __name__ == "__main__":
    main()

# HTTP REST Client with Retry Logic (Python - Trial 1)

An HTTP REST client featuring automatic retry with exponential backoff, request chaining, response validation, and structured logging.

## Dependencies

- **requests** (2.31.0) - HTTP library for Python
- **tenacity** (8.2.3) - Retry library with configurable backoff strategies

## Features

- GET, POST, PUT, PATCH, DELETE request methods
- Automatic retry with exponential backoff for transient errors
- Configurable retry status codes (default: 429, 500, 502, 503, 504)
- Request chaining with context interpolation between steps
- Response validation with built-in and custom rules
- Full request/response history with structured logging

## Usage

```bash
pip install -r requirements.txt
python client.py
```

## API

```python
from client import RestClient, status_ok, json_has_key

client = RestClient(
    base_url="https://api.example.com",
    max_retries=3,
    backoff_multiplier=1.0,
)

# Simple GET
response = client.get("/endpoint", validators=[status_ok()])

# Request chaining
from client import RequestConfig
configs = [
    RequestConfig(method="GET", url="/users/1", label="get_user"),
    RequestConfig(method="GET", url="/users/{user_id}/posts", label="get_posts"),
]
results = client.chain(configs, extractor=my_extractor)
```

requests==2.31.0
tenacity==8.2.3