starlette-dial/pkg/dial

Dial - HTTP 客户端和代理工具

Dial 是一个基于 httpx 的异步 HTTP 客户端类，提供了连接池管理、直接请求和代理功能。

功能特性

• 🔄 连接池管理: 自动管理 HTTP 连接池，提高性能
• 🌐 直接请求: 支持各种 HTTP 方法的直接请求
• 🔀 代理功能: 可以创建代理处理器，支持请求和响应重写
• ⚡ 异步支持: 完全异步实现，与 Starlette 框架完美集成
• 🛡️ 类型安全: 完整的类型注解支持
• 🚨 异常处理: 内置 HTTP 状态码异常类，自动处理错误响应

安装依赖

确保已安装 httpx 依赖：

pip install httpx

或在 pyproject.toml 中添加：

dependencies = [
    "httpx>=0.27.0",
]

基本使用

1. 直接请求

from pkg.dial import client

# 简单 GET 请求
response = await client.dial("https://api.example.com/users")
print(response.status_code)
print(response.json())

# POST 请求带 JSON 数据
response = await client.dial(
    url="https://api.example.com/users",
    method="POST",
    headers={"Content-Type": "application/json"},
    body={"name": "John", "email": "john@example.com"}
)

2. 异常处理

from pkg.dial import client, BadRequestException, ServerErrorException

try:
    response = await client.dial("https://api.example.com/users")
except BadRequestException as e:
    print(f"请求参数错误: {e.msg}")
    print(f"错误详情: {e.err}")
    print(f"返回数据: {e.data}")
except ServerErrorException as e:
    print(f"服务器错误: {e.msg}")

3. 自定义客户端

from pkg.dial import Dial
import httpx

# 创建自定义客户端
dial = Dial(
    timeout=30.0,
    limits=httpx.Limits(max_keepalive_connections=10, max_connections=50)
)

# 使用上下文管理器
async with dial as client:
    response = await client.dial("https://api.example.com/data")
    print(response.text)

4. 代理功能

from pkg.dial import client
from starlette.requests import Request
from starlette.responses import JSONResponse

# 请求处理函数
def request_handler(request: Request):
    return {
        'url': f'https://api.example.com{request.url.path}',
        'method': request.method,
        'headers': dict(request.headers),
        'params': dict(request.query_params)
    }

# 响应处理函数
def response_handler(response):
    return JSONResponse({
        'status': response.status_code,
        'data': response.json()
    })

# 创建代理处理器
proxy_handler = client.proxy(
    req_fn=request_handler,
    res_fn=response_handler
)

# 在 Starlette 应用中使用
from starlette.applications import Starlette
from starlette.routing import Route

app = Starlette(routes=[
    Route('/api/{path:path}', proxy_handler)
])

API 参考

Dial 类

构造函数

Dial(timeout: float = 30.0, limits: Optional[httpx.Limits] = None)

• timeout: 请求超时时间（秒）
• limits: HTTP 连接限制配置

方法

dial()

async def dial(
    self, 
    url: str, 
    method: str = "GET", 
    headers: Optional[Dict[str, str]] = None, 
    body: Optional[Any] = None,
    params: Optional[Dict[str, Any]] = None
) -> httpx.Response

直接发送 HTTP 请求。当响应状态码 >= 300 时，会自动抛出相应的异常。

proxy()

def proxy(
    self, 
    req_fn: Optional[Callable[[Request], Dict[str, Any]]] = None, 
    res_fn: Optional[Callable[[httpx.Response], Response]] = None
) -> Callable[[Request], Response]

创建代理处理器。

• req_fn: 请求处理函数，接收 Request 对象，返回请求配置字典
• res_fn: 响应处理函数，接收 httpx.Response 对象，返回 Response 对象

close()

async def close()

关闭客户端连接。

异常类

BadRequestException

400 错误异常，用于处理请求参数错误。

class BadRequestException(Exception):
    def __init__(self, msg="", data=None, err=None, status=None):
        self.status = status or 400
        self.msg = msg or "参数错误"
        self.data = data
        self.err = err

UnauthorizationException

401 错误异常，用于处理认证失败。

class UnauthorizationException(Exception):
    def __init__(self, msg="", data=None, err=None, status=None):
        self.status = status or 401
        self.msg = msg or "登录信息不存在或已过期, 请重新登录"
        self.data = data
        self.err = err

ForbiddenException

403 错误异常，用于处理权限不足。

class ForbiddenException(Exception):
    def __init__(self, msg="", data=None, err=None, status=None):
        self.status = status or 403
        self.msg = msg or "权限不足"
        self.data = data
        self.err = err

NotFoundException

404 错误异常，用于处理资源不存在。

class NotFoundException(Exception):
    def __init__(self, msg="", data=None, err=None, status=None):
        self.status = status or 404
        self.msg = msg or "资源不存在"
        self.data = data
        self.err = err

ServerErrorException

500 错误异常，用于处理服务器内部错误。

class ServerErrorException(Exception):
    def __init__(self, msg="", data=None, err=None, status=None):
        self.status = status or 500
        self.msg = msg or "服务器开小差了"
        self.data = data
        self.err = err

异常处理器

为了方便在 Starlette 应用中使用，还提供了对应的异常处理器：

from pkg.dial.exception import (
    bad_request_exception_handler,
    unauthorization_exception_handler,
    forbidden_exception_handler,
    not_found_exception_handler,
    server_error_exception_handler,
)

# 在 Starlette 应用中注册异常处理器
app.add_exception_handler(BadRequestException, bad_request_exception_handler)
app.add_exception_handler(UnauthorizationException, unauthorization_exception_handler)
app.add_exception_handler(ForbiddenException, forbidden_exception_handler)
app.add_exception_handler(NotFoundException, not_found_exception_handler)
app.add_exception_handler(ServerErrorException, server_error_exception_handler)

示例

查看 examples/dial_example.py 文件获取完整的使用示例。

注意事项

1. 连接池管理: 客户端会自动管理连接池，无需手动管理
2. 异步使用: 所有方法都是异步的，需要在异步环境中使用
3. 资源清理: 使用上下文管理器或手动调用 close() 方法清理资源
4. 类型安全: 建议使用类型注解以获得更好的开发体验