py_easy_httpx

一个简化 httpx 库使用的 Python 工具包，提供同步和异步 HTTP 请求的便捷封装，以及响应处理功能。

功能特性

• 同步/异步支持：同时支持同步和异步 HTTP 请求，分别通过 HttpxRequest 和 AsyncHttpxRequest 类实现
• 灵活配置：支持通过参数自定义客户端行为
• 响应处理：提供 JSON 数据验证和提取功能
• JSON Schema 验证：支持使用 JSON Schema 验证响应数据
• JSONPath 支持：支持使用 JSONPath 提取特定数据
• 类型提示：完整的类型注解，提供良好的 IDE 支持

安装

使用 pip 安装

pip install py_easy_httpx

从源码安装

git clone https://gitee.com/guolei19850528/py_easy_httpx.git
cd py_easy_httpx
pip install -e .

依赖

• httpx
• decorator
• jsonschema
• jsonpath-ng

快速开始

基本用法

同步请求

from py_easy_httpx.request import HttpxRequest

# 初始化请求实例
request = HttpxRequest(client_kwargs={"timeout": 10.0})

# 发送 GET 请求
response = request.request(method="GET", url="https://httpbin.org/get")
print(response.status_code)
print(response.json())

# 发送 POST 请求
response = request.request(method="POST", url="https://httpbin.org/post", json={"name": "test"})
print(response.status_code)
print(response.json())

# 使用自定义客户端
client = request.client(timeout=60.0, base_url="https://httpbin.org")
try:
    # 发送请求
    response = request.request(client=client, method="GET", url="/get")
    print(response.status_code)
finally:
    # 手动关闭客户端
    client.close()

异步请求

import asyncio
from py_easy_httpx.request import AsyncHttpxRequest

async def main():
    # 初始化异步请求实例
    request = AsyncHttpxRequest(client_kwargs={"timeout": 10.0})
    
    # 发送异步 GET 请求
    response = await request.async_request(method="GET", url="https://httpbin.org/get")
    print(response.status_code)
    print(response.json())

asyncio.run(main())

使用响应处理装饰器

同步装饰器

from py_easy_httpx.request import HttpxRequest
from py_easy_httpx.response import json_paser

# 初始化请求实例
request = HttpxRequest()

# 使用装饰器验证响应并提取数据
@json_paser(
    validator={"type": "object", "properties": {"args": {"type": "object"}}},
    path_expr="$.args"
)
def get_data():
    response = request.request(method="GET", url="https://httpbin.org/get", params={"page": 1, "limit": 10})
    return response

# 获取提取的数据
data = get_data()
print(f"提取的数据: {data}")

异步装饰器

import asyncio
from py_easy_httpx.request import AsyncHttpxRequest
from py_easy_httpx.response import async_json_paser

async def main():
    # 初始化异步请求实例
    request = AsyncHttpxRequest()
    
    # 使用异步装饰器验证响应并提取数据
    @async_json_paser(
        validator={"type": "object", "properties": {"json": {"type": "object"}}},
        path_expr="$.json"
    )
    async def post_data():
        response = await request.async_request(method="POST", url="https://httpbin.org/post", json={"name": "test", "value": 123})
        return response
    
    # 获取提取的数据
    data = await post_data()
    print(f"提取的数据: {data}")

asyncio.run(main())

API 文档

HttpxRequest 类

初始化参数

• client_kwargs：传递给 httpx.Client 的默认参数

方法

• client(**kwargs)：创建同步客户端实例
• request(client=None, **kwargs)：同步发送请求

AsyncHttpxRequest 类

初始化参数

• client_kwargs：传递给 httpx.AsyncClient 的默认参数

方法

• async_client(**kwargs)：创建异步客户端实例
• async_request(client=None, **kwargs)：异步发送请求

响应处理装饰器

json_paser(validator=None, path_expr=None)

同步响应处理装饰器

• 参数:
  • validator：JSON Schema 验证规则
  • path_expr：JSONPath 表达式，用于提取数据
• 返回：装饰器函数，装饰后的函数会返回：
  • 如果 Schema 验证失败：None
  • 如果 Schema 验证通过且未指定 path_expr：True
  • 如果 Schema 验证通过且指定了 path_expr：提取的第一个匹配值

async_json_paser(validator=None, path_expr=None)

异步响应处理装饰器

• 参数:
  • validator：JSON Schema 验证规则
  • path_expr：JSONPath 表达式，用于提取数据
• 返回：装饰器函数，装饰后的函数会返回：
  • 如果 Schema 验证失败：None
  • 如果 Schema 验证通过且未指定 path_expr：True
  • 如果 Schema 验证通过且指定了 path_expr：提取的第一个匹配值

高级用法

自定义客户端配置

from py_easy_httpx.request import HttpxRequest, AsyncHttpxRequest

# 初始化同步请求实例，配置默认客户端参数
sync_request = HttpxRequest(
    client_kwargs={
        "timeout": 30.0,
        "base_url": "https://api.example.com",
        "headers": {
            "User-Agent": "py_easy_httpx",
            "Authorization": "Bearer your_token_here"
        }
    }
)

# 发送同步请求，使用默认配置
response = sync_request.request(method="GET", url="/users")
print(response.status_code)

# 初始化异步请求实例
async_request = AsyncHttpxRequest(client_kwargs={"timeout": 30.0})

# 创建临时异步客户端，覆盖部分默认配置
async_client = async_request.async_client(timeout=60.0)
try:
    # 发送异步请求
    response = await async_request.async_request(client=async_client, method="GET", url="https://api.example.com/products")
    print(response.status_code)
finally:
    # 手动关闭异步客户端
    await async_client.aclose()

响应数据提取

from py_easy_httpx.request import HttpxRequest
from py_easy_httpx.response import json_paser

request = HttpxRequest()

# 提取嵌套字段
@json_paser(
    validator={
        "type": "object",
        "properties": {
            "data": {
                "type": "object",
                "properties": {
                    "user": {
                        "type": "object",
                        "properties": {
                            "id": {"type": "integer"},
                            "name": {"type": "string"}
                        }
                    }
                }
            }
        }
    },
    path_expr="$.data.user.name"
)
def get_username():
    # 模拟 API 响应
    class MockResponse:
        def json(self):
            return {
                "data": {
                    "user": {
                        "id": 1,
                        "name": "John Doe"
                    }
                }
            }
    return MockResponse()

username = get_username()
print(f"用户名: {username}")

项目结构

py_easy_httpx/
├── py_easy_httpx/              # 主包目录
│   ├── __init__.py            # 包初始化文件
│   ├── request.py             # 请求处理模块（包含 HttpxRequest 和 AsyncHttpxRequest 类）
│   └── response.py            # 响应处理模块
├── README.md                  # 项目文档
├── setup.py                   # 安装配置
├── requirements.txt           # 依赖列表
├── LICENSE                    # 许可证文件
├── deploy.sh                  # 部署脚本
└── .gitignore                 # Git 忽略文件

测试

运行测试：

# 运行测试
python -m pytest

许可证

MIT License

贡献

欢迎提交 Issue 和 Pull Request！

联系方式

• 作者：guolei
• 邮箱：174000902@qq.com
• 项目地址：https://gitee.com/guolei19850528/py_easy_httpx

致谢

• httpx - 高性能的异步 HTTP 客户端库
• jsonschema - JSON Schema 验证库
• jsonpath-ng - JSONPath 实现库