Python网络请求还在用requests?httpx才是现代开发的首选
不知道你们有没有这种感觉,用requests写接口调用代码,写着写着总觉得哪里不对劲。
我去年接了一个数据采集的项目,需要从七八个不同的API获取数据,然后用asyncio并发处理。一开始用requests,结果发现这玩意儿是同步的,想并发就得手动开线程池。后来换了aiohttp,API风格突然变了,大量回调嵌套,代码读起来像在走迷宫。
直到我发现了httpx。
httpx是什么?简单说就是一个同时支持同步和异步的HTTP客户端库。它有requests的简洁API,又能跑出aiohttp的速度。最关键的是,它让我终于不用在两个完全不同的写代码风格之间来回切换了。
01.
一、为什么你需要httpx
先说说requests的问题。
requests的口碑一直很好,API设计优雅,文档详细,生态成熟。我用了requests好几年,从来没想过换它。但直到我开始写异步代码,才发现它的局限性。
问题在于requests是纯同步的。想象一下这个场景:你要从5个不同的API获取数据,每个API响应时间是200毫秒。用requests同步调用,总耗时是5乘以200,等于1秒。但如果这5个请求之间没有依赖关系,为什么要排队等呢?
我第一次意识到这个问题,是帮朋友优化一个爬虫脚本。他的代码用requests循环调用了20个API,跑了将近40秒。我把循环改成并发,耗时直接降到3秒左右。整整十几倍的差距,就因为从串行变成了并行。
这就是httpx的核心价值:它让同步和异步代码可以用同一套API写。你不需要学两套完全不同的用法。
02.
二、5分钟快速上手
先安装httpx:
BASHpip install httpx
发送GET请求
这是最基础的操作:
PYTHONimport httpx# 同步方式response = httpx.get("https://api.github.com/users/octocat")print(response.status_code)print(response.json())
这段代码看起来和requests一模一样。httpx的设计哲学就是这样——对于简单场景,你不需要改变任何习惯。
发送POST请求
POST请求同样简单:
PYTHON# 发送JSON数据response = httpx.post( "https://httpbin.org/post", json={"name": "test", "value": 123})print(response.json())# 发送表单数据response = httpx.post( "https://httpbin.org/post", data={"username": "admin", "password": "123456"})
注意到没有?json=参数直接传字典,httpx会自动帮你序列化成JSON,还加上Content-Type头。同样的,用data=发送表单数据,Content-Type也会自动变成application/x-www-form-urlencoded。
我第一次用的时候觉得这功能太贴心了。之前用requests,还要手动json.dumps(),还要设置header,现在一行搞定。
处理响应
httpx的响应对象提供了丰富的方法:
PYTHONresponse = httpx.get("https://httpbin.org/json")# 获取文本内容print(response.text)# 获取JSON(自动解析)data = response.json()# 获取二进制内容(图片、文件等)content = response.content# 检查状态码if response.status_code == 200: print("请求成功")
这里有个细节值得注意:response.json()方法会自动解析响应体,如果响应不是合法的JSON格式,会抛出异常。我的建议是加上异常处理:
PYTHONtry: data = response.json()except ValueError: print("响应不是JSON格式") data = {}
03.
三、请求的高级配置
自定义请求头
有些场景需要自定义请求头,比如模拟浏览器访问:
PYTHONheaders = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "application/json", "Authorization": "Bearer your-token-here"}response = httpx.get("https://api.example.com/data", headers=headers)
我之前踩过一个坑。写爬虫的时候,用requests设置了User-Agent,但服务器返回403。一查才发现,请求头里还有个Accept-Encoding没设置,导致服务端认为是爬虫。httpx默认会加上常见头,减少了这种低级错误。
URL参数处理
Query参数不用自己拼接字符串:
PYTHONparams = { "page": 1, "limit": 20, "category": "tech", "sort": "-created_at"}response = httpx.get("https://api.example.com/articles", params=params)# 实际URL: https://api.example.com/articles?page=1&limit=20&category=tech&sort=-created_at
这个功能虽然简单,但真的很实用。手动拼接URL参数容易出错,还要处理特殊字符的URL编码,httpx全帮你搞定了。
超时控制
网络请求必须设置超时,否则一旦网络出问题,你的程序可能永远卡在那里。
PYTHON# 设置全局超时(所有请求)client = httpx.Client(timeout=10.0)# 不同类型操作设置不同超时client = httpx.Client( timeout=httpx.Timeout( connect=5.0, # 连接超时 read=30.0, # 读取超时 write=10.0, # 写入超时 pool=5.0 # 池化超时 ))response = client.get("https://api.example.com/data")
我曾经写过一个定时任务,调用外部API获取数据。有一段时间API服务不稳定,经常超时。结果定时任务开始堆积,最后服务器内存耗尽。从那以后,我给所有网络请求都加上超时控制,这是血泪教训。
04.
四、异步模式:真正的高并发
这是httpx的杀手锏。
基本异步请求
PYTHONimport httpximport asyncioasync def fetch_data(): async with httpx.AsyncClient() as client: response = await client.get("https://api.github.com/users/octocat") return response.json()# 运行异步函数data = asyncio.run(fetch_data())print(data)
语法看起来有点怪,但理解了就很简单。async def定义异步函数,await等待异步操作完成,AsyncClient是异步版本的客户端。
并发请求
这才体现出异步的威力:
PYTHONimport httpximport asyncioasync def fetch_all(): urls = [ "https://api.github.com/users/octocat", "https://api.github.com/users/torvalds", "https://api.github.com/users/pallets", "https://api.github.com/users/psf", ] async with httpx.AsyncClient() as client: # 并发发送所有请求 responses = await asyncio.gather( *[client.get(url) for url in urls] ) # 解析所有响应 data = [r.json() for r in responses] return dataresults = asyncio.run(fetch_all())print(f"获取了 {len(results)} 个用户信息")
asyncio.gather是关键函数。它会并发执行所有传入的协程,等全部完成后再返回结果。实测中,4个URL并发请求耗时约等于最慢那个请求的耗时,而不是4个请求耗时的总和。
异步客户端配置
异步客户端支持配置连接池,更好地控制并发行为:
PYTHONasync def fetch_with_pool(): # 配置连接池限制 limits = httpx.Limits( max_keepalive_connections=20, # 最大保活连接数 max_connections=100 # 最大连接总数 ) async with httpx.AsyncClient(limits=limits, timeout=30.0) as client: # 批量请求 tasks = [client.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return [r.json() for r in responses]
连接池配置在高并发场景下很重要。如果不做限制,瞬间发起几千个请求可能导致服务器拒绝连接或被封IP。
05.
五、会话管理:连接复用
如果你需要多次请求同一个域名,用会话可以复用TCP连接,大幅提升性能:
PYTHON# 同步版本with httpx.Client() as client: # 基础URL client.base_url = "https://api.github.com" # 自动保持连接 response1 = client.get("/users/octocat") response2 = client.get("/users/torvalds") response3 = client.get("/users/pallets")# 异步版本async def fetch_user_data(): async with httpx.AsyncClient(base_url="https://api.github.com") as client: response1 = await client.get("/users/octocat") response2 = await client.get("/users/torvalds")
会话的好处不只是复用连接。如果你的API需要认证,可以在客户端层面设置通用的请求头,不用每个请求都重复写:
PYTHONwith httpx.Client( base_url="https://api.example.com", headers={"Authorization": "Bearer token123"}) as client: # 所有请求自动带上Authorization头 r1 = client.get("/protected/resource1") r2 = client.get("/protected/resource2")
06.
六、实战案例:封装API客户端
聊完基础用法,来点实战。
我写API调用代码有个习惯,喜欢封装成类。这样做有几个好处:配置集中管理、接口统一、方便测试和复用。
封装GitHub API客户端
PYTHONimport httpxfrom typing import Optional, List, Dictclass GitHubClient: """GitHub API客户端封装""" def __init__(self, token: Optional[str] = None): self.base_url = "https://api.github.com" headers = { "Accept": "application/vnd.github.v3+json", "User-Agent": "httpx-demo-client" } if token: headers["Authorization"] = f"Bearer {token}" self.client = httpx.Client(base_url=self.base_url, headers=headers) def get_user(self, username: str) -> Dict: """获取用户信息""" response = self.client.get(f"/users/{username}") response.raise_for_status() return response.json() def get_repos(self, username: str, sort: str = "updated") -> List[Dict]: """获取用户的仓库列表""" response = self.client.get( f"/users/{username}/repos", params={"sort": sort, "per_page": 100} ) response.raise_for_status() return response.json() def search_repos(self, keyword: str, language: Optional[str] = None) -> List[Dict]: """搜索仓库""" params = {"q": keyword} if language: params["q"] += f" language:{language}" response = self.client.get("/search/repositories", params=params) response.raise_for_status() return response.json()["items"] def close(self): """关闭客户端""" self.client.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close()# 使用示例with GitHubClient() as client: # 获取用户信息 user = client.get_user("octocat") print(f"用户名: {user['login']}") print(f"粉丝数: {user['followers']}") # 获取最近更新的仓库 repos = client.get_repos("octocat", sort="pushed") for repo in repos[:5]: print(f" {repo['name']}: {repo['description']}") # 搜索Python项目 results = client.search_repos("web framework", language="Python") print(f"\n找到 {len(results)} 个Python Web框架")
这段代码展示了封装的思想。API客户端把认证、基础URL、通用配置都封装在初始化方法里,对外只暴露业务方法。调用方不需要关心底层实现,只需要调用get_user()、get_repos()这样的高层接口。
异步版本API客户端
PYTHONimport httpximport asynciofrom typing import Optional, List, Dictclass AsyncGitHubClient: """异步版本GitHub API客户端""" def __init__(self, token: Optional[str] = None): self.base_url = "https://api.github.com" headers = { "Accept": "application/vnd.github.v3+json", "User-Agent": "httpx-async-demo" } if token: headers["Authorization"] = f"Bearer {token}" self.client = httpx.AsyncClient( base_url=self.base_url, headers=headers, timeout=30.0 ) async def get_user(self, username: str) -> Dict: response = await self.client.get(f"/users/{username}") response.raise_for_status() return response.json() async def get_repos(self, username: str, sort: str = "updated") -> List[Dict]: response = await self.client.get( f"/users/{username}/repos", params={"sort": sort, "per_page": 100} ) response.raise_for_status() return response.json() async def get_multiple_users(self, usernames: List[str]) -> List[Dict]: """并发获取多个用户信息""" tasks = [self.get_user(username) for username in usernames] return await asyncio.gather(*tasks) async def close(self): await self.client.aclose() async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.close()# 使用示例async def main(): async with AsyncGitHubClient() as client: # 串行获取(慢) # user1 = await client.get_user("octocat") # user2 = await client.get_user("torvalds") # 并发获取(快) users = await client.get_multiple_users([ "octocat", "torvalds", "pallets", "psf", "microsoft" ]) print(f"并发获取了 {len(users)} 个用户信息") for user in users: print(f" {user['login']}: {user['followers']} followers")asyncio.run(main())
异步版本的客户端用起来更灵活。get_multiple_users方法内部使用asyncio.gather并发请求多个用户信息,这在需要批量获取数据时特别有用。
07.
七、httpx vs requests vs aiohttp
最后来个横向对比,帮你们做选择。
| 特性 | httpx | requests | aiohttp |
|------|-------|----------|---------|
| 同步支持 | ✅ | ✅ | ❌ |
| 异步支持 | ✅ | ❌ | ✅ |
| API风格 | 简洁现代 | 简洁经典 | 回调式 |
| 类型提示 | 完整 | 无 | 部分 |
| 并发性能 | 优秀 | 一般 | 优秀 |
| 依赖大小 | 轻量 | 轻量 | 中等 |
我的选择建议:
如果你只写同步代码,而且代码已经稳定运行,不改也行。requests够用。
如果你的项目需要异步支持,httpx是最佳选择。一套API走天下,不需要在requests和aiohttp之间切换。
如果你要新项目,直接上httpx。它兼顾了易用性和性能,没有任何理由不选它。
08.
八、踩坑实录
说几个我踩过的坑。
坑1:忘记关闭异步客户端
PYTHON# 错误写法async def bad_example(): client = httpx.AsyncClient() response = await client.get("https://example.com") # 函数结束,client没有被关闭 # 连接泄漏!# 正确写法async def good_example(): async with httpx.AsyncClient() as client: response = await client.get("https://example.com") # 自动关闭,不会泄漏
坑2:异常处理不当
PYTHON# 错误写法:不检查状态码response = httpx.get("https://api.example.com/data")print(response.json()) # 服务器500错误也会继续执行# 正确写法:抛出异常response = httpx.get("https://api.example.com/data")response.raise_for_status() # 4xx/5xx响应会抛出异常data = response.json()# 更健壮的写法try: response = httpx.get("https://api.example.com/data") response.raise_for_status() data = response.json()except httpx.HTTPStatusError as e: print(f"HTTP错误: {e.response.status_code}")except httpx.RequestError as e: print(f"请求错误: {e}")
坑3:异步函数写成同步
PYTHON# 错误写法async def fetch_data(): client = httpx.AsyncClient() response = client.get("https://example.com") # 缺少await! return response.json()# 正确写法async def fetch_data(): async with httpx.AsyncClient() as client: response = await client.get("https://example.com") return response.json()
这个坑我踩过两次。忘记写await的时候,代码不会报错,但返回的是协程对象而不是响应结果。调试了很久才发现问题出在这么简单的语法上。
09.
九、写在最后
httpx不是银弹,不是说用了它你的代码就能脱胎换骨。但它确实解决了一个真实的痛点:同步和异步的统一。
我用过requests很多年,也用过aiohttp写爬虫。两种体验完全不同,直到httpx出现。同一套API,同一种写代码的方式,同步异步无缝切换。这种感觉,就像从手动挡换到了自动挡——不是说手动挡不好,只是自动挡让驾驶变得更简单了。
建议你们有机会试试httpx,不用急着重构现有代码,先在新项目里用用看。体验一下它带来的便利,再决定要不要全面迁移。
📌 更多Python技术干货,关注"Python与AI智能研习社"~