Boost Your API Test Automation with a Powerful Python HTTP Client

Why deep encapsulation is needed for API automation?

Directly using requests leads to repetitive code, tangled assertion logic, unclear failure messages, noisy logs, missing retry/global headers/environment switching, and unstructured reports.

Core encapsulation: HttpClient with automation features

Install required dependencies: pip install requests colorama colorlog Complete wrapper code (ready to use):

import requests
import logging
from typing import Any, Dict, Optional
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from colorama import init, Fore, Style

init(autoreset=True)  # Enable Windows colored output

# =======================
# 🎨 Colored print function
# =======================

def cprint(text: str, color: str = "white", style: str = "normal"):
    colors = {
        "black": Fore.BLACK, "red": Fore.RED, "green": Fore.GREEN,
        "yellow": Fore.YELLOW, "blue": Fore.BLUE, "magenta": Fore.MAGENTA,
        "cyan": Fore.CYAN, "white": Fore.WHITE,
        "bright_red": Fore.LIGHTRED_EX, "bright_green": Fore.LIGHTGREEN_EX,
        "bright_yellow": Fore.LIGHTYELLOW_EX, "bright_blue": Fore.LIGHTBLUE_EX
    }
    styles = {"normal": Style.NORMAL, "bright": Style.BRIGHT, "dim": Style.DIM}
    color_code = colors.get(color.lower(), Fore.WHITE)
    style_code = styles.get(style.lower(), Style.NORMAL)
    print(f"{style_code}{color_code}{text}{Style.RESET_ALL}")

# =======================
# 🧪 Assertion utility
# =======================
class Assert:
    def __init__(self, response: Dict):
        self.resp = response
        self.success = True
        self.errors = []

    def status(self, expected: int):
        actual = self.resp.get("status_code")
        if actual != expected:
            msg = f"Status code assertion failed: expected {expected}, actual {actual}"
            self.errors.append(msg)
            cprint(msg, "red")
            self.success = False
        else:
            cprint(f"✅ Status code {actual} correct", "green")
        return self

    def json(self, key: str, expected: Any):
        data = self.resp.get("data", {})
        actual = data.get(key)
        if actual != expected:
            msg = f"JSON assertion failed: {key} expected '{expected}', actual '{actual}'"
            self.errors.append(msg)
            cprint(msg, "red")
            self.success = False
        else:
            cprint(f"✅ JSON field '{key}': {actual}", "green")
        return self

    def contains(self, substring: str):
        text = self.resp.get("data", {}).get("raw_text", "")
        if substring not in text:
            msg = f"Response does not contain: '{substring}'"
            self.errors.append(msg)
            cprint(msg, "yellow")
            self.success = False
        else:
            cprint(f"✅ Response contains '{substring}'", "green")
        return self

# =======================
# 🚀 Automation‑focused HTTP client
# =======================
class ApiClient:
    def __init__(self, base_url: str, timeout: int = 10, retries: int = 2, headers: Optional[Dict[str, str]] = None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(headers or {})
        self._setup_retry(retries)
        self._setup_logger()

    def _setup_retry(self, retries: int):
        retry = Retry(total=retries, backoff_factor=0.3, status_forcelist=[500, 502, 503, 504])
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def _setup_logger(self):
        logging.basicConfig(level=logging.INFO, format="%(message)s", handlers=[logging.StreamHandler()])

    def _request(self, method: str, url: str, **kwargs) -> Dict[str, Any]:
        full_url = self.base_url + ("" if url.startswith("/") else "/") + url
        start_time = __import__('time').time()
        cprint(f"🚀 {method} {full_url}", "bright_blue", "bright")
        try:
            response = self.session.request(method, full_url, timeout=self.timeout, **kwargs)
            response.raise_for_status()
            duration = round((__import__('time').time() - start_time) * 1000, 2)
            try:
                data = response.json()
            except Exception:
                data = {"raw_text": response.text[:500]}
            cprint(f"✅ [{response.status_code}] Response time: {duration}ms", "bright_green")
            result = {
                "success": True,
                "status_code": response.status_code,
                "data": data,
                "headers": dict(response.headers),
                "url": response.url,
                "duration_ms": duration,
                "timestamp": __import__('time').strftime("%Y-%m-%d %H:%M:%S")
            }
            return result
        except Exception as e:
            duration = round((__import__('time').time() - start_time) * 1000, 2)
            cprint(f"❌ [{getattr(e.response, 'status_code', 'N/A')}] Request failed: {e}", "bright_red")
            return {
                "success": False,
                "error": str(e),
                "url": full_url,
                "duration_ms": duration,
                "timestamp": __import__('time').strftime("%Y-%m-%d %H:%M:%S")
            }

    def get(self, url, **kwargs):
        return Assert(self._request("GET", url, **kwargs))

    def post(self, url, **kwargs):
        return Assert(self._request("POST", url, **kwargs))

    def put(self, url, **kwargs):
        return Assert(self._request("PUT", url, **kwargs))

    def delete(self, url, **kwargs):
        return Assert(self._request("DELETE", url, **kwargs))

Usage example: one‑line request + assertion

# test_user.py
# Initialize client (configurable for different environments)
client = ApiClient(
    base_url="https://jsonplaceholder.typicode.com",
    headers={"Authorization": "Bearer fake-token"}
)
# Test 1: Get user info
client.get("/posts/1") \
    .status(200) \
    .json("userId", 1) \
    .json("title", "sunt aut facere repellat provident occaecati excepturi optio reprehenderit")
# Test 2: Create a user
payload = {"title": "自动化测试", "body": "这是一条测试数据", "userId": 1}
client.post("/posts", json=payload) \
    .status(201) \
    .json("title", "自动化测试")

Running result (colored terminal output)

🚀 GET https://jsonplaceholder.typicode.com/posts/1

✅ [200] Response time: 123.45ms

✅ Status code 200 correct

✅ JSON field 'userId': 1

✅ JSON field 'title': sunt aut facere...

🚀 POST https://jsonplaceholder.typicode.com/posts

✅ [201] Response time: 234.12ms

✅ Status code 201 correct

✅ JSON field 'title': 自动化测试