Python Decorators for API Testing: Retry, Timeout, Logging, Caching, Validation, and More

Decorators are special Python functions that can modify the behavior of other functions; they are applied by prefixing a function definition with @decorator_name . The article presents a series of useful decorators for API testing.

Retry decorator allows automatic re‑execution of a failing request up to a configurable number of attempts with a delay between tries.

import requests
from functools import wraps
import time

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise e
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=1)
def send_request(url):
    """发送 GET 请求并返回响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

result = send_request("https://api.example.com/data")
print("响应内容:", result)  # 响应内容: {'key': 'value'}

Timeout decorator limits the execution time of a function using the signal module.

import requests
import signal
from functools import wraps

class TimeoutError(Exception):
    pass

def timeout(seconds):
    def decorator(func):
        def handler(signum, frame):
            raise TimeoutError(f"Timeout after {seconds} seconds")
        @wraps(func)
        def wrapper(*args, **kwargs):
            old_handler = signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            try:
                return func(*args, **kwargs)
            finally:
                signal.alarm(0)
                signal.signal(signal.SIGALRM, old_handler)
        return wrapper
    return decorator

@timeout(5)
def send_request(url):
    """发送 GET 请求并返回响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

try:
    result = send_request("https://api.example.com/data")
    print("响应内容:", result)
except TimeoutError as e:
    print(e)  # Timeout after 5 seconds

Logging decorator prints the request URL and response status code.

import requests
from functools import wraps

def log_request(func):
    @wraps(func)
    def wrapper(url, *args, **kwargs):
        print("发送请求:", url)
        response = func(url, *args, **kwargs)
        print("响应状态码:", response.status_code)
        return response
    return wrapper

@log_request
def send_request(url):
    """发送 GET 请求并返回响应"""
    return requests.get(url)

response = send_request("https://api.example.com/data")
print("响应内容:", response.json())  # 响应内容: {'key': 'value'}

Caching decorator uses functools.lru_cache to cache results of API calls.

import requests
from functools import lru_cache

@lru_cache(maxsize=32)
def get_user(id):
    """根据用户ID获取用户信息"""
    response = requests.get(f"https://api.example.com/user/{id}")
    response.raise_for_status()
    return response.json()

user = get_user(1)
print("用户:", user)  # 用户: {'id': 1, 'name': 'User 1'}

Response validation decorator asserts that the HTTP status is 200 and that a required key exists in the JSON payload.

import requests
from functools import wraps

def validate_response(func):
    @wraps(func)
    def wrapper(url, *args, **kwargs):
        response = func(url, *args, **kwargs)
        assert response.status_code == 200, "响应状态码错误"
        assert "key" in response.json(), "响应中缺少关键字段"
        return response
    return wrapper

@validate_response
def send_request(url):
    """发送 GET 请求并返回响应"""
    return requests.get(url)

response = send_request("https://api.example.com/data")
print("响应内容:", response.json())  # 响应内容: {'key': 'value'}

Parameterization decorator runs a test function for each set of parameters supplied.

import requests
from functools import wraps

def parametrize(params):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for param in params:
                func(param)
        return wrapper
    return decorator

@parametrize([{"url": "https://api.example.com/data1"}, {"url": "https://api.example.com/data2"}])
def send_request(url):
    """发送 GET 请求并打印响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    print("响应内容:", response.json())

send_request()
# 输出：
# 响应内容: {'key': 'value1'}
# 响应内容: {'key': 'value2'}

Exception handling decorator catches requests exceptions and returns None instead of propagating the error.

import requests
from functools import wraps

def handle_exception(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.RequestException as e:
            print(f"请求异常: {e}")
            return None
    return wrapper

@handle_exception
def send_request(url):
    """发送 GET 请求并返回响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

result = send_request("https://api.example.com/data")
if result:
    print("响应内容:", result)  # 响应内容: {'key': 'value'}

Performance monitoring decorator measures and prints the execution time of a function.

import requests
import time
from functools import wraps

def monitor_performance(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        elapsed_time = time.time() - start_time
        print(f"函数执行耗时: {elapsed_time:.2f} 秒")
        return result
    return wrapper

@monitor_performance
def send_request(url):
    """发送 GET 请求并返回响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

result = send_request("https://api.example.com/data")
print("响应内容:", result)  # 响应内容: {'key': 'value'}

Permission checking decorator verifies that the caller has an allowed permission before executing the function.

from functools import wraps

def check_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if permission not in ["admin", "user"]:
                raise PermissionError("没有权限执行该操作")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@check_permission("admin")
def delete_user(id):
    """删除用户"""
    print(f"删除用户 {id}")

delete_user(1)  # 删除用户 1

Composite decorator demonstrates how multiple decorators can be stacked to combine logging and timeout functionality.

import requests
from functools import wraps

def log_request(func):
    @wraps(func)
    def wrapper(url, *args, **kwargs):
        print("发送请求:", url)
        return func(url, *args, **kwargs)
    return wrapper

def timeout(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

@log_request
@timeout(5)
def send_request(url):
    """发送 GET 请求并返回响应内容"""
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

result = send_request("https://api.example.com/data")
print("响应内容:", result)  # 响应内容: {'key': 'value'}

By employing these Python decorators, API automation test code becomes more concise, robust, and maintainable, allowing developers to focus on test logic rather than repetitive boilerplate.