# py_easy_httpx **Repository Path**: guolei19850528/py_easy_httpx ## Basic Information - **Project Name**: py_easy_httpx - **Description**: py_easy_httpx - **Primary Language**: Python - **License**: MIT - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-02-09 - **Last Updated**: 2026-02-14 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # 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 实现库