Python脚本命令模式如何封装请求

wen 实用脚本 2026-07-19 5

本文目录导读：

Python脚本命令模式如何封装请求

基础命令模式实现
增强版命令模式
使用示例
配置文件支持

我来详细介绍Python脚本中封装请求的命令模式实现方案。

基础命令模式实现

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
import json
import requests
from dataclasses import dataclass, field
from enum import Enum
# 请求方法枚举
class HttpMethod(Enum):
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"
# 命令接口
class Command(ABC):
    @abstractmethod
    def execute(self) -> Any:
        pass
    @abstractmethod
    def undo(self) -> None:
        pass
# 请求数据封装
@dataclass
class RequestData:
    url: str
    method: HttpMethod
    headers: Dict[str, str] = field(default_factory=dict)
    params: Dict[str, Any] = field(default_factory=dict)
    data: Optional[Dict[str, Any]] = None
    json_data: Optional[Dict[str, Any]] = None
    timeout: int = 30
    def to_dict(self) -> Dict[str, Any]:
        return {
            'url': self.url,
            'method': self.method.value,
            'headers': self.headers,
            'params': self.params,
            'data': self.data,
            'json': self.json_data,
            'timeout': self.timeout
        }
# 响应数据封装
@dataclass
class ResponseData:
    status_code: int
    headers: Dict[str, str]
    content: bytes
    elapsed: float
    def json(self) -> Dict:
        return json.loads(self.content)
    def text(self) -> str:
        return self.content.decode('utf-8')
# 具体命令类
class HttpRequestCommand(Command):
    def __init__(self, request_data: RequestData):
        self.request_data = request_data
        self.response = None
        self.session = requests.Session()
    def execute(self) -> ResponseData:
        try:
            request_dict = self.request_data.to_dict()
            method = request_dict.pop('method')
            # 发送请求
            response = self.session.request(method, **request_dict)
            # 封装响应
            self.response = ResponseData(
                status_code=response.status_code,
                headers=dict(response.headers),
                content=response.content,
                elapsed=response.elapsed.total_seconds()
            )
            return self.response
        except requests.RequestException as e:
            raise CommandExecutionError(f"请求执行失败: {str(e)}")
    def undo(self) -> None:
        """撤销操作 - 对于HTTP请求，可以记录日志或清理资源"""
        if self.response:
            print(f"撤销请求: {self.request_data.method.value} {self.request_data.url}")
            self.response = None
# 自定义异常
class CommandExecutionError(Exception):
    pass
# 命令调用者
class RequestInvoker:
    def __init__(self):
        self._command = None
        self._history = []
    def set_command(self, command: Command):
        self._command = command
    def execute_command(self) -> Any:
        if not self._command:
            raise ValueError("未设置命令")
        result = self._command.execute()
        self._history.append(self._command)
        return result
    def undo_last_command(self) -> None:
        if self._history:
            command = self._history.pop()
            command.undo()
    def clear_history(self) -> None:
        self._history.clear()

增强版命令模式

import asyncio
from typing import List, Callable, Optional
from functools import wraps
import time
import logging
# 日志装饰器
def log_execution(func: Callable):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"执行命令: {func.__name__}")
        start_time = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start_time
        logging.info(f"命令完成: {func.__name__}, 耗时: {elapsed:.2f}s")
        return result
    return wrapper
# 重试机制
def retry(max_retries: int = 3, delay: float = 1.0):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))
                        logging.warning(f"重试 {attempt + 1}/{max_retries}")
            raise last_exception
        return wrapper
    return decorator
# 高级请求命令
class AdvancedHttpCommand(Command):
    def __init__(self, request_data: RequestData, 
                 max_retries: int = 3,
                 on_success: Optional[Callable] = None,
                 on_error: Optional[Callable] = None):
        self.request_data = request_data
        self.max_retries = max_retries
        self.on_success = on_success
        self.on_error = on_error
        self.response = None
        self._retry_count = 0
    @log_execution
    @retry(max_retries=3)
    def execute(self) -> ResponseData:
        try:
            response = self._send_request()
            self.response = response
            # 调用成功回调
            if self.on_success:
                self.on_success(response)
            return response
        except Exception as e:
            # 调用错误回调
            if self.on_error:
                self.on_error(e)
            raise
    def _send_request(self) -> ResponseData:
        session = requests.Session()
        request_dict = self.request_data.to_dict()
        method = request_dict.pop('method')
        response = session.request(method, **request_dict)
        response.raise_for_status()  # 检查HTTP状态码
        return ResponseData(
            status_code=response.status_code,
            headers=dict(response.headers),
            content=response.content,
            elapsed=response.elapsed.total_seconds()
        )
    def undo(self) -> None:
        if self.response:
            # 实现撤销逻辑，比如记录日志或缓存清理
            logging.info(f"撤销请求: {self.request_data.url}")
            self.response = None
# 命令组合器
class CompositeCommand(Command):
    def __init__(self, commands: List[Command] = None):
        self.commands = commands or []
    def add_command(self, command: Command):
        self.commands.append(command)
    def remove_command(self, command: Command):
        self.commands.remove(command)
    def execute(self) -> List[Any]:
        results = []
        for command in self.commands:
            result = command.execute()
            results.append(result)
        return results
    def undo(self) -> None:
        for command in reversed(self.commands):
            command.undo()
# 异步命令
class AsyncHttpCommand(Command):
    def __init__(self, request_data: RequestData):
        self.request_data = request_data
        self.response = None
    async def execute_async(self) -> ResponseData:
        import aiohttp
        async with aiohttp.ClientSession() as session:
            async with session.request(
                self.request_data.method.value,
                self.request_data.url,
                headers=self.request_data.headers,
                params=self.request_data.params,
                json=self.request_data.json_data
            ) as response:
                content = await response.read()
                self.response = ResponseData(
                    status_code=response.status,
                    headers=dict(response.headers),
                    content=content,
                    elapsed=0  # async方式需要额外计算
                )
                return self.response
    def execute(self) -> ResponseData:
        # 同步方式的回调
        return asyncio.run(self.execute_async())
    def undo(self) -> None:
        self.response = None
# 命令工厂
class CommandFactory:
    @staticmethod
    def create_get_request(url: str, params: Dict = None, headers: Dict = None) -> HttpRequestCommand:
        request_data = RequestData(
            url=url,
            method=HttpMethod.GET,
            params=params or {},
            headers=headers or {}
        )
        return HttpRequestCommand(request_data)
    @staticmethod
    def create_post_request(url: str, json_data: Dict, headers: Dict = None) -> HttpRequestCommand:
        request_data = RequestData(
            url=url,
            method=HttpMethod.POST,
            json_data=json_data,
            headers=headers or {}
        )
        return HttpRequestCommand(request_data)
    @staticmethod
    def create_advanced_command(request_data: RequestData, **kwargs) -> AdvancedHttpCommand:
        return AdvancedHttpCommand(request_data, **kwargs)

使用示例

def example_usage():
    # 1. 基本使用
    invoker = RequestInvoker()
    # 创建GET请求
    get_command = CommandFactory.create_get_request(
        url="https://api.example.com/users",
        params={"page": 1},
        headers={"Authorization": "Bearer token123"}
    )
    invoker.set_command(get_command)
    try:
        response = invoker.execute_command()
        print(f"状态码: {response.status_code}")
        print(f"响应内容: {response.text()}")
    except CommandExecutionError as e:
        print(f"错误: {e}")
    # 2. 高级使用
    request_data = RequestData(
        url="https://api.example.com/users",
        method=HttpMethod.POST,
        json_data={"name": "John", "age": 30},
        headers={"Content-Type": "application/json"}
    )
    def on_success(response):
        print(f"请求成功: {response.status_code}")
    def on_error(error):
        print(f"请求失败: {error}")
    advanced_command = CommandFactory.create_advanced_command(
        request_data,
        max_retries=3,
        on_success=on_success,
        on_error=on_error
    )
    invoker.set_command(advanced_command)
    result = invoker.execute_command()
    # 3. 批量请求
    composite = CompositeCommand()
    # 添加多个请求
    commands = [
        CommandFactory.create_get_request("https://api.example.com/endpoint1"),
        CommandFactory.create_get_request("https://api.example.com/endpoint2"),
        CommandFactory.create_get_request("https://api.example.com/endpoint3")
    ]
    for cmd in commands:
        composite.add_command(cmd)
    # 执行批量请求
    results = composite.execute()
    # 4. 撤销操作
    invoker.undo_last_command()

配置文件支持

import yaml
from pathlib import Path
# 配置加载
class CommandConfig:
    def __init__(self, config_path: str = "requests.yaml"):
        self.config = self._load_config(config_path)
    def _load_config(self, path: str) -> Dict:
        with open(path, 'r') as f:
            return yaml.safe_load(f)
    def create_commands_from_config(self) -> List[Command]:
        commands = []
        for request_config in self.config.get('requests', []):
            request_data = RequestData(
                url=request_config['url'],
                method=HttpMethod[request_config['method']],
                headers=request_config.get('headers', {}),
                params=request_config.get('params', {}),
                json_data=request_config.get('json_data'),
                timeout=request_config.get('timeout', 30)
            )
            commands.append(HttpRequestCommand(request_data))
        return commands
# 示例YAML配置文件
"""
requests:
  - url: "https://api.example.com/users"
    method: GET
    headers:
      Authorization: "Bearer token"
    params:
      page: 1
      limit: 10
  - url: "https://api.example.com/users"
    method: POST
    json_data:
      name: "John"
      age: 30
    timeout: 10
"""

核心优势：

解耦：请求的发送者与接收者完全解耦
可扩展：易于添加新的命令类型
可回滚：支持撤销操作
可组合：支持命令组合和批处理

适用场景：

需要记录请求历史
需要支持请求重试
需要批量处理请求
需要支持事务性操作

这个模式特别适合API客户端、爬虫系统、微服务调用等需要处理大量HTTP请求的场景。