当前位置:首页>python>Python HTTP新选择:httpx一站式解决所有网络请求

Python HTTP新选择:httpx一站式解决所有网络请求

  • 2026-10-11 08:29:22
Python HTTP新选择:httpx一站式解决所有网络请求

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智能研习社"~

最新文章

随机文章