如何基于openapi-python-client实现Python客户端会话复用与连接池优化
复用openapi-python-client生成的客户端会话最佳实践
核心思路
openapi-python-client生成的Client(同步)和AsyncClient(异步)类支持传入自定义的httpx.Client/httpx.AsyncClient实例,这类实例自带连接池管理和SSL会话复用能力,所有请求都会复用同一个会话,无需重复处理SSL或创建新连接。这是无需修改生成代码的最优方案。
1. 同步客户端会话复用
实现步骤
- 创建自定义的
httpx.Client实例,配置SSL验证、超时等全局参数 - 将该实例传入生成的
Client类的构造方法 - 后续所有API调用自动复用这个会话
示例代码
import httpx from your_generated_client import Client # 创建带连接池的httpx客户端,复用SSL会话 custom_httpx_client = httpx.Client( verify=True, # 对应原client.verify_ssl配置 timeout=30.0, # 可添加全局headers、代理等其他配置 ) # 初始化生成的Client,传入自定义httpx客户端 api_client = Client(base_url="https://your-api-server.com", httpx_client=custom_httpx_client) # 调用API,自动复用会话 response = api_client.sync_detailed(service="your-service", all_fields=True) # 使用完毕后关闭会话 custom_httpx_client.close()
2. 异步客户端会话复用
修正你提供的异步代码错误
你的异步代码存在语法问题,正确实现需通过await调用异步方法,并正确获取响应:
async def asyncio_detailed( *, client: AsyncClient, # 注意此处应为生成的AsyncClient类 all_fields: Union[Unset, None, bool] = False, service: str, ) -> Response[Res]: kwargs = _get_kwargs( client=client, all_fields=all_fields, service=service, ) # 调用异步客户端的request方法,必须加await response = await client.httpx_client.request( verify=client.verify_ssl, **kwargs, ) return _build_response(response=response)
异步会话复用示例
import httpx import asyncio from your_generated_client import AsyncClient async def main(): # 创建异步httpx客户端 custom_async_client = httpx.AsyncClient( verify=True, timeout=30.0, ) # 初始化生成的AsyncClient api_client = AsyncClient(base_url="https://your-api-server.com", httpx_client=custom_async_client) # 异步调用API,复用会话 response = await api_client.asyncio_detailed(service="your-service", all_fields=True) # 关闭异步会话 await custom_async_client.aclose() asyncio.run(main())
3. 自定义模板方案(可选)
如果需要永久修改生成代码的请求逻辑,可通过自定义模板实现:
- 利用openapi-python-client的自定义模板路径功能,修改同步/异步请求的模板文件
- 将模板中的
httpx.request替换为client.httpx_client.request(同步)或await client.httpx_client.request(异步) - 重新生成客户端代码,后续生成的代码都会默认复用会话
注意:该方案需要维护自定义模板,适合长期固定需求场景,否则优先推荐传入自定义客户端的方式。
内容的提问来源于stack exchange,提问作者Ihab Ame
相关产品推荐
相关产品推荐

