Master Python API Automation with 10 Essential Decorators

When writing API automation tests, merely sending requests with requests is insufficient; real projects encounter timeouts, token handling, slow responses, and flaky tests. Decorators provide a clean way to encapsulate repetitive logic.

What is a decorator?

A decorator is a wrapper function that adds behavior before and after the original function call. The example below demonstrates a simple @say_hello decorator that prints messages around the execution of add:

def say_hello(func):
    def wrapper(*args, **kwargs):
        print("开始执行")
        result = func(*args, **kwargs)
        print("执行完毕")
        return result
    return wrapper

@say_hello
def add(a, b):
    return a + b

add(1, 2)

Output:

开始执行
执行完毕

10 Ready‑to‑Use Decorators

1. Automatic Retry @retry

Scenario: An order API occasionally returns 500 due to network jitter. The decorator retries the request up to max_times with a pause of interval seconds.

import time, functools

def retry(max_times=3, interval=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"第 {attempt} 次执行失败：{e}")
                    if attempt < max_times:
                        time.sleep(interval)
                    else:
                        raise
        return wrapper
    return decorator

@retry(max_times=3, interval=2)
def test_order_api():
    resp = requests.post("http://example.com/api/order", json={"item_id": 1})
    assert resp.status_code == 200, f"接口返回异常：{resp.status_code}"

test_order_api()

After applying @retry, a failed call automatically waits 2 seconds and retries up to three times, keeping the test code tidy.

2. Execution Timing @timeit

Scenario: Measure how long a login API takes.

import time, functools

def timeit(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f"[耗时] {func.__name__} 执行了 {end - start:.3f} 秒")
        return result
    return wrapper

@timeit
def test_login_api():
    resp = requests.post("http://example.com/api/login", json={"username": "test_user", "password": "123456"})
    assert resp.status_code == 200

test_login_api()

Output example: [耗时] test_login_api 执行了 0.213 秒.

3. Token Injection @with_token

Scenario: Many endpoints require an Authorization: Bearer … header. The decorator fetches a token once and injects it into each test.

import functools, requests
_token_cache = {}

def get_token():
    if "token" not in _token_cache:
        resp = requests.post("http://example.com/api/login", json={"username": "auto_test", "password": "test123"})
        _token_cache["token"] = resp.json()["token"]
    return _token_cache["token"]

def with_token(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        kwargs["token"] = get_token()
        return func(*args, **kwargs)
    return wrapper

@with_token
def test_get_user_info(token):
    headers = {"Authorization": f"Bearer {token}"}
    resp = requests.get("http://example.com/api/user/info", headers=headers)
    assert resp.status_code == 200
    assert resp.json()["username"] == "auto_test"

test_get_user_info()

The login logic lives only in get_token(), and the token is cached for reuse.

4. Detailed Failure Logging @log_on_failure

Scenario: When a test fails, you want the full stack trace and request details.

import functools, traceback

def log_on_failure(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print("=" * 50)
            print(f"用例失败：{func.__name__}")
            print(f"错误信息：{e}")
            print(f"完整堆栈：
{traceback.format_exc()}")
            print("=" * 50)
            raise
    return wrapper

@log_on_failure
def test_create_user():
    resp = requests.post("http://example.com/api/user", json={"name": "张三", "email": "invalid-email"})
    assert resp.status_code == 201, f"实际响应：{resp.status_code}, body: {resp.text}"

test_create_user()

The decorator prints a clear separator, the test name, the exception, and the full traceback before re‑raising.

5. Skip in Specific Environments @skip_in_env

Scenario: Destructive tests should not run in production or staging.

import os, functools

def skip_in_env(*envs):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            current_env = os.getenv("TEST_ENV", "dev")
            if current_env in envs:
                print(f"[跳过] {func.__name__} 在 {current_env} 环境下不执行")
                return None
            return func(*args, **kwargs)
        return wrapper
    return decorator

@skip_in_env("prod", "staging")
def test_delete_all_test_data():
    resp = requests.delete("http://example.com/api/testdata/clear")
    assert resp.status_code == 200

os.environ["TEST_ENV"] = "prod"
test_delete_all_test_data()

Output:

[跳过] test_delete_all_test_data 在 prod 环境下不执行

.

6. Response Schema Validation @validate_schema

Scenario: An API returns 200 but the JSON structure is wrong.

import functools

def validate_schema(expected_keys):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            resp = func(*args, **kwargs)
            if resp is None:
                return resp
            body = resp.json()
            missing = [k for k in expected_keys if k not in body]
            if missing:
                raise AssertionError(f"响应缺少字段：{missing}，实际返回：{list(body.keys())}")
            print(f"[校验通过] {func.__name__} 响应结构正常")
            return resp
        return wrapper
    return decorator

@validate_schema(expected_keys=["id", "username", "email", "created_at"])
def test_get_user_detail():
    resp = requests.get("http://example.com/api/user/1")
    assert resp.status_code == 200
    return resp

test_get_user_detail()

All required fields are maintained in one place; updating the API doc only requires editing the decorator argument.

7. Execution Timeout @timeout

Scenario: Prevent a test from hanging indefinitely.

import functools, threading

def timeout(seconds=10):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = [None]
            exception = [None]
            def target():
                try:
                    result[0] = func(*args, **kwargs)
                except Exception as e:
                    exception[0] = e
            thread = threading.Thread(target=target)
            thread.daemon = True
            thread.start()
            thread.join(timeout=seconds)
            if thread.is_alive():
                raise TimeoutError(f"{func.__name__} 执行超时（>{seconds}秒）")
            if exception[0]:
                raise exception[0]
            return result[0]
        return wrapper
    return decorator

@timeout(seconds=5)
def test_slow_api():
    import time
    time.sleep(8)  # Simulate a slow endpoint
    resp = requests.get("http://example.com/api/report/generate")
    assert resp.status_code == 200

test_slow_api()

Result: TimeoutError: test_slow_api 执行超时（>5秒）.

8. Automatic Cleanup @cleanup

Scenario: Tests that create data must delete it afterwards.

import functools

def cleanup(cleanup_func):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            created_ids = []
            try:
                result = func(*args, created_ids=created_ids, **kwargs)
                return result
            finally:
                if created_ids:
                    cleanup_func(created_ids)
        return wrapper
    return decorator

def delete_test_users(user_ids):
    print(f"清理测试用户：{user_ids}")
    for uid in user_ids:
        requests.delete(f"http://example.com/api/user/{uid}")

@cleanup(cleanup_func=delete_test_users)
def test_create_multiple_users(created_ids):
    for i in range(3):
        resp = requests.post("http://example.com/api/user", json={"name": f"测试用户{i}"})
        assert resp.status_code == 201
        created_ids.append(resp.json()["id"])

test_create_multiple_users()

The finally block guarantees cleanup even if the test fails.

9. Full HTTP Recording @record_http

Scenario: Capture request and response details for every test case.

import functools, json
from unittest.mock import patch
http_records = []

def record_http(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        original_request = requests.Session.request
        def patched_request(self, method, url, **req_kwargs):
            resp = original_request(self, method, url, **req_kwargs)
            record = {
                "test_case": func.__name__,
                "method": method.upper(),
                "url": url,
                "request_body": req_kwargs.get("json") or req_kwargs.get("data"),
                "status_code": resp.status_code,
                "response_body": resp.text[:500]
            }
            http_records.append(record)
            print(f"[HTTP记录] {method.upper()}{url} -> {resp.status_code}")
            return resp
        with patch.object(requests.Session, "request", patched_request):
            return func(*args, **kwargs)
    return wrapper

@record_http
def test_user_workflow():
    requests.post("http://example.com/api/login", json={"username": "u1", "password": "p1"})
    requests.get("http://example.com/api/user/info")

test_user_workflow()
print("
完整 HTTP 记录:")
print(json.dumps(http_records, ensure_ascii=False, indent=2))

The decorator uses unittest.mock.patch to intercept all HTTP calls without changing test code.

10. Quick Concurrent Stress Test @concurrent

Scenario: Verify an endpoint under ten simultaneous requests without a full load‑testing tool.

import functools, threading

def concurrent(users=10):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            errors = []
            lock = threading.Lock()
            def run():
                try:
                    func(*args, **kwargs)
                except Exception as e:
                    with lock:
                        errors.append(str(e))
            threads = [threading.Thread(target=run) for _ in range(users)]
            for t in threads:
                t.start()
            for t in threads:
                t.join()
            if errors:
                raise AssertionError(f"{len(errors)}/{users} 个并发请求失败：
" + "
".join(errors[:5]))
            print(f"[并发测试] {users} 个并发全部通过")
            return None
        return wrapper
    return decorator

@concurrent(users=10)
def test_search_api_concurrent():
    resp = requests.get("http://example.com/api/search", params={"q": "python"})
    assert resp.status_code == 200
    assert "results" in resp.json()

test_search_api_concurrent()

Any failing thread is reported, and the overall test fails if any request errors.

Combining Decorators

Decorators can be stacked; the outermost runs first. Example combining timing, retry, and logging:

@timeit
@retry(max_times=3, interval=1)
@log_on_failure
def test_payment_api():
    resp = requests.post("http://example.com/api/pay", json={"order_id": "123"})
    assert resp.status_code == 200

test_payment_api()

Execution order: timeit starts timing → retry handles retries → log_on_failure logs any exception.

Conclusion

The ten decorators presented solve concrete pain points encountered in everyday API test automation. They are ready‑to‑copy snippets; adapt the URLs, payloads, and field names to your own project and enjoy cleaner, more maintainable test code.