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

# py_easy_httpx

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

## 功能特性

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

## 安装

### 使用 pip 安装

```bash
pip install py_easy_httpx
```

### 从源码安装

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

## 依赖

- httpx
- decorator
- jsonschema
- jsonpath-ng

## 快速开始

### 基本用法

#### 同步请求

```python
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()
```

#### 异步请求

```python
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())
```

### 使用响应处理装饰器

#### 同步装饰器

```python
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}")
```

#### 异步装饰器

```python
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：提取的第一个匹配值

## 高级用法

### 自定义客户端配置

```python
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()
```

### 响应数据提取

```python
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 忽略文件
```

## 测试

运行测试：

```bash
# 运行测试
python -m pytest
```

## 许可证

MIT License

## 贡献

欢迎提交 Issue 和 Pull Request！

## 联系方式

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

## 致谢

- [httpx](https://www.python-httpx.org/) - 高性能的异步 HTTP 客户端库
- [jsonschema](https://python-jsonschema.readthedocs.io/) - JSON Schema 验证库
- [jsonpath-ng](https://github.com/h2non/jsonpath-ng) - JSONPath 实现库