python第三方网络库httpx

字数统计: 1.6k字   |   阅读时长: 7分

HTTPX是Python3的全功能HTTP客户端,它提供同步和异步API,并支持HTTP/1.1和HTTP/2。

介绍

根据官网的描述，总结有如下特点:

  • 和使用 requests 一样方便,requests 有的它都有
  • 加入 HTTP/1.1 和 HTTP/2 的支持。
  • 能够直接向 WSGI 应用程序或 ASGI 应用程序发出请求。
  • 到处都有严格的超时设置
  • 全类型注释
  • 100％ 的测试覆盖率

github介绍:https://github.com/encode/httpx

文档介绍:https://www.python-httpx.org/

安装

httpx安装很简单，直接使用pip安装即可

1
pip install httpx

基本使用

httpx的使用和requests类似，可以直接使用get/post/put/delete等方法发送请求

下面就是简单实用httpx发送GET/POST/PUT/DELETE/PATCH/HEAD/OPTIONS请求的例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import httpx

# 本地调试后端地址
base_url = "http://10.209.0.23:5100/api/model"


def test_get():
    """
    测试get请求
    get请求和requests差不多，也支持代理模式、重定向、证书认证等
    """
    url = base_url
    response = httpx.get(url)
    print(response.text)
    print(response.json())
    print(response.status_code)


def test_post():
    """
    测试post请求
    post请求，同样支持json、formdata、files等类型的数据
    """
    url = base_url
    body = {
        "name": "test",
        "detail": "test",
    }
    response = httpx.post(url, json=body)
    print(response.text)
    print(response.status_code)


def test_put():
    """
    测试put请求
    """
    url = f"{base_url}/123687577743360"
    body = {
        "name": "test-update",
        "detail": "test-update",
        "level": 123123,
    }
    response = httpx.put(url, json=body)
    print(response.text)
    print(response.status_code)


def test_delete():
    """
    测试delete请求
    """
    url = f"{base_url}/123687577743360"
    response = httpx.delete(url)
    print(response.text)
    print(response.status_code)


def test_patch():
    """
    测试patch请求
    """
    url = f"{base_url}/123687577743360"
    body = {
        "name": "test-patch",
    }
    response = httpx.patch(url, json=body)
    print(response.text)
    print(response.status_code)


def test_head():
    """
    测试head请求
    """
    url = base_url
    response = httpx.head(url)
    print(response.headers)
    print(response.status_code)


def test_options():
    """
    测试options请求
    """
    url = base_url
    response = httpx.options(url)
    print(response.headers)
    print(response.status_code)
    print(response.headers.get("allow"))
    

if __name__ == '__main__':
    test_get()
    # test_post()
    # test_put()
    # test_delete()
    # test_patch()
    test_head()
    test_options()

高级使用

如果我们直接跳转上面的请求方法可以看到，httpx的请求方法最终都是通过使用Client类来实现的，所以我们可以直接使用Client类来实现更多的功能

httpx.Client类支持同步和异步请求，支持tcp连接池、代理、重定向、证书认证、cookies等功能

Client类使用方式

Client类的使用方式有两种，一种是作为上下文管理器，一种是直接实例化。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import httpx

# 作为上下文管理器使用，会自动关闭连接
with httpx.Client() as client:
    response = client.get("https://www.baidu.com")
    print(response.text)
    
# 直接实例化，需要手动关闭连接
client = httpx.Client()
try:
    response = client.get("https://www.baidu.com")
    print(response.text)
finally:
    client.close()

base_url

设置base_url是一个比较基础的功能，可以在实例化Client类的时候设置base_url，这样在请求的时候就不需要每次都写完整的url了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from httpx import Client


def test_get(httpx_client: Client, url: str = "/"):
    """
    测试get请求
    """
    response = httpx_client.get(url)
    print(response.text)
    print(response.json())
    print(response.status_code)


if __name__ == '__main__':
    with Client(base_url="http://10.209.0.23:5100/api/") as client:
        # 模拟三次请求
        test_get(client, url="model")
        test_get(client, url="model")
        test_get(client, url="model")

异步请求

httpx支持异步请求，使用异步请求可以提高性能，但是需要注意的是，异步请求的时候，需要使用async with来作为上下文管理器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

from httpx import AsyncClient


async def test_get(httpx_client: AsyncClient, url: str = "/"):
    """
    测试get请求
    """
    response = await httpx_client.get(url)
    print(response.text)
    print(response.json())
    print(response.status_code)


async def main():
    async with AsyncClient(base_url="http://10.209.0.23:5100/api/") as client:
        await test_get(client, url="model")


if __name__ == '__main__':
    asyncio.run(main())

连接池

httpx支持tcp连接池，可以提高性能，使用方式也很简单，只需要在实例化Client类的时候设置连接池的大小即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import asyncio

from httpx import AsyncClient, Limits


async def test_get(httpx_client: AsyncClient, url: str = "/"):
    """
    测试get请求
    """
    response = await httpx_client.get(url)
    print(response.text)
    print(response.json())
    print(response.status_code)


async def main():
    # 创建一个最大连接数为100，最大保持活动连接数为20的连接池
    limits = Limits(max_connections=100, max_keepalive_connections=20)
    # 应用连接池
    async with AsyncClient(base_url="http://10.209.0.23:5100/api/", limits=limits) as client:
        await test_get(client, url="model")


if __name__ == '__main__':
    asyncio.run(main())

钩子函数

httpx支持钩子函数，可以在请求前后执行一些操作，比如请求前打印日志，请求后打印日志等

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import asyncio

from httpx import AsyncClient, Limits, Request, Response


async def log_request(request: Request):
    """
    请求钩子，用于打印请求信息
    :param request: 请求对象
    :return:
    """
    print(f">>> 发送请求: {request.method} {request.url}")
    print(">>> 请求头:")
    for name, value in request.headers.items():
        print(f"{name}: {value}")
    print(">>> 请求体:")
    print(request.content)


async def log_response(response: Response):
    """
    响应钩子，用于打印响应信息
    :param response: 响应对象
    :return:
    """
    print(f"<<< 接收响应: {response.status_code} {response.url}")
    print("<<< 响应头:")
    for name, value in response.headers.items():
        print(f"{name}: {value}")

    # 读取响应体，由于响应体是异步的，所以需要使用aread()方法。aread()方法只能调用一次，多次调用会抛出异常
    # 需要再单独执行一次读取操作的原因是响应体一般比较大，如果在响应钩子中读取，会导致响应钩子执行时间过长，影响性能，因此需要显示的调用aread()方法
    # 注意：只有在响应钩子中读取响应体时，才需要使用aread()方法，其他情况下，可以直接使用response.text、response.json()等方法读取
    content = await response.aread()
    print("<<< 响应体:")
    print(content)


async def test_get(httpx_client: AsyncClient, url: str = "/"):
    """
    测试get请求
    """
    response = await httpx_client.get(url)
    print(response.text)
    print(response.json())
    print(response.status_code)


async def main():
    # 创建一个最大连接数为100，最大保持活动连接数为20的连接池
    limits = Limits(max_connections=100, max_keepalive_connections=20)
    # 应用连接池
    async with AsyncClient(
            base_url="http://10.209.0.23:5100/api/",
            limits=limits,
            event_hooks={
                # 注册请求钩子
                "request": [log_request],
                # 注册响应钩子
                "response": [log_response]
            }
    ) as client:
        await test_get(client, url="model")


if __name__ == '__main__':
    asyncio.run(main())

本文标题:python第三方网络库httpx

文章作者:aaron

发布时间:2023-12-26, 11:47:01

最后更新:2023-12-26, 16:34:34

原始链接:http://realtiger.github.io/python-httpx/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。

文章目录
  1. 1. 介绍
  2. 2. 安装
  3. 3. 基本使用
  4. 4. 高级使用
    1. 4.1. Client类使用方式
    2. 4.2. base_url
    3. 4.3. 异步请求
    4. 4.4. 连接池
    5. 4.5. 钩子函数