Py学习  »  Python

httpx,一个超酷的 Python 库!

python • 1 年前 • 316 次点击  


大家好,今天为大家分享一个超酷的 Python 库 - httpx。

Github地址:https://github.com/encode/httpx


httpx是一个现代化的Python HTTP客户端库,它在保持与requests库相似API设计的基础上,增加了对异步编程的全面支持。作为requests库的精神继承者,httpx不仅提供了同步的HTTP请求功能,还原生支持async/await语法。

httpx专门为现代Python应用程序设计,特别适合需要处理大量并发HTTP请求的场景,如API集成、网络爬虫、微服务通信等。httpx的设计理念是提供一个功能完整、性能优异且易于使用的HTTP客户端,让开发者能够轻松处理各种复杂的网络请求需求。

安装

1、基础安装

使用pip命令可以轻松安装httpx库:




    
pip install httpx

如果需要使用HTTP/2功能,建议安装完整版本:

pip install httpx[http2]

2、验证安装

安装完成后,可以通过以下代码验证是否安装成功:

import httpx
print(httpx.__version__)

# 简单测试请求
response = httpx.get('https://httpbin.org/json')
print(response.status_code)

特性

支持同步和异步两种编程模式,开发者可以根据项目需求灵活选择。提供与requests库高度兼容的API接口,降低学习成本和迁移难度。原生支持HTTP/2协议,提供更好的网络性能和连接复用能力。

内置连接池管理机制,自动处理连接的创建、复用和释放。支持完整的认证体系,包括基础认证、摘要认证和自定义认证方案。提供强大的会话管理功能,支持Cookie持久化和请求配置共享。

具备完善的超时控制机制,支持连接超时、读取超时等多种超时类型。支持代理服务器配置,包括HTTP和SOCKS代理。提供详细的请求和响应信息,便于调试和监控。

基本功能

1、基础HTTP请求

httpx提供了简洁的API来执行各种HTTP请求方法。GET请求用于获取服务器资源,POST请求用于提交数据到服务器,PUT和DELETE请求分别用于更新和删除资源。

import httpx

# GET请求
response = httpx.get('https://api.github.com/users/octocat')
print(response.json())

# POST请求提交JSON数据
data = {'name''John''age'30}
response = httpx.post('https://httpbin.org/post', json=data)
print(response.status_code)

# PUT请求更新资源
response = httpx.put('https://httpbin.org/put' , data={'key''value'})

# DELETE请求删除资源
response = httpx.delete('https://httpbin.org/delete')

2、异步HTTP请求

异步请求是httpx的核心优势之一,特别适合需要同时处理多个HTTP请求的场景。使用async/await语法可以显著提高程序的并发性能,避免因网络IO阻塞而导致的性能瓶颈。异步客户端需要在异步上下文中使用,通过上下文管理器确保资源的正确释放。

import asyncio
import httpx

asyncdef fetch_data():
    asyncwith httpx.AsyncClient() as client:
        # 异步GET请求
        response = await client.get('https://api.github.com/users/octocat')
        return response.json()

# 并发请求多个URL
asyncdef fetch_multiple():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1'
    ]
    
    asyncwith httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.status_code for r in responses]

# 运行异步函数
result = asyncio.run(fetch_data())
print(result)

高级功能

1、会话管理和配置

httpx的Client类提供了强大的会话管理功能,允许在多个请求之间共享配置、Cookie和连接池。

import httpx

# 创建会话客户端
with httpx.Client() as client:
    # 设置默认头部
    client.headers.update({'User-Agent''MyApp/1.0'})
    
    # 登录获取认证信息
    login_response = client.post('https://httpbin.org/post'
                                json={'username''user''password''pass'})
    
    # 使用会话状态进行后续请求
    protected_response = client.get('https://httpbin.org/headers')
    print(protected_response.json())

2、超时和重试机制

在生产环境中,网络请求的超时控制至关重要。httpx提供了灵活的超时配置选项,支持连接超时、读取超时和总体超时的精确控制。合理的超时设置可以防止程序因网络问题而长时间阻塞,提高系统的稳定性和响应能力。

import httpx
from httpx import Timeout

# 配置超时时间
timeout = Timeout(10.0, connect=5.0)

with httpx.Client(timeout=timeout) as client:
    try:
        response = client.get('https://httpbin.org/delay/3')
        print(f"请求成功: {response.status_code}")
    except httpx.TimeoutException:
        print("请求超时")
    except httpx.HTTPError as e:
        print(f"HTTP错误: {e}")

实际应用场景

1、API数据聚合服务

在实际项目中,经常需要从多个API源获取数据并进行聚合处理。使用httpx的异步功能可以显著提高数据获取效率,特别是在需要调用多个外部服务的场景中。

import asyncio
import httpx

class DataAggregator:
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=30.0)
    
    asyncdef fetch_user_info(self, user_id):
        response = await self.client.get(f'https://api.example.com/users/{user_id}')
        return response.json()
    
    asyncdef fetch_user_posts(self, user_id):
        response = await self.client.get(f'https://api.example.com/users/{user_id}/posts')
        return response.json()
    
    asyncdef get_complete_user_data(self, user_id):
        user_info, user_posts = await asyncio.gather(
            self.fetch_user_info(user_id),
            self.fetch_user_posts(user_id)
        )
        return {'info': user_info, 'posts': user_posts}
    
    asyncdef close(self):
         await self.client.aclose()

2、网络监控和健康检查

httpx非常适合构建服务监控系统,可以定期检查多个服务的可用性状态,及时发现和报告系统问题。

import asyncio
import httpx
from datetime import datetime

asyncdef health_check():
    services = [
        'https://api.service1.com/health',
        'https://api.service2.com/health',
        'https://api.service3.com/health'
    ]
    
    asyncwith httpx.AsyncClient(timeout=5.0as client:
        for service in services:
            try:
                response = await client.get(service)
                status = "健康"if response.status_code == 200else"异常"
                print(f"{datetime.now()} - {service}{status}")
            except Exception as e:
                print(f"{datetime.now()} - {service}: 连接失败 - {e}")

# 定期执行健康检查
asyncio.run(health_check())

总结

httpx不仅保持了与requests库相似的API设计,降低了开发者的学习成本,更重要的是通过原生支持async/await语法,为构建高并发网络应用提供了强有力的技术支撑。核心优势体现在其全面的功能支持和优秀的性能表现上。从基础的HTTP请求操作到复杂的会话管理,从精确的超时控制到高效的连接池管理,httpx都提供了解决方案。

如果你觉得文章还不错,请大家 点赞、分享、留言 下,因为这将是我持续输出更多优质文章的最强动力!

我们还为大家准备了Python资料,感兴趣的小伙伴快来找我领取一起交流学习哦!

图片

往期推荐

历时一个月整理的 Python 爬虫学习手册全集PDF(免费开放下载)

Beautiful Soup快速上手指南,从入门到精通(PDF下载)

Python基础学习常见的100个问题.pdf(附答案)

124个Python案例,完整源代码!

30 个Python爬虫的实战项目(附源码)

从入门到入魔,100个Python实战项目练习(附答案)!

80个Python数据分析必备实战案例.pdf(附代码),完全开放下载

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/182682