Python网络请求-httpx

发表于 | 分类于
Python网络请求-httpx

前言

在 Python 众多的 HTTP 客户端中，最有名的莫过于requests、aiohttp和httpx。

在不借助其他第三方库的情况下，requests只能发送同步请求；aiohttp只能发送异步请求；httpx既能发送同步请求，又能发送异步请求。

那么怎么选择呢

  • 只发同步请求用requests，但可配合多线程变异步。
  • 只发异步请求用aiohttp，但可以配合await变同步。
  • httpx可以发同步请求也可以异步，但是请求速度同步略差于requests，异步略差于aiohttp

这里不建议使用多线程来做异步请求，建议使用异步IO的方式。

httpx的特点：

  • 功能强大，既能同步也能异步。
  • 同步请求和requests的语法基本一致，方便代码迁移。
  • 性能虽然差点，但是差的不多可以忽略。

同步请求

GET请求

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

r = httpx.get(
    'https://www.psvmc.cn/login.json',
    params={'keyword': '123'}
)

print("r.text:", r.text)
print("r.json():", r.json())
print("r.content:", r.content)

# 响应码
print("r.status_code:", r.status_code)
# 响应的编码
print("r.encoding:", r.encoding)
# 请求的URL
print("r.url:", r.url)
# Cookie
print("r.cookies:", r.cookies)
# 响应的header
print("r.headers:", r.headers)
print("r.headers['Content-Type']:", r.headers['Content-Type'])
# 请求的header
print("r.request.headers:", r.request.headers)

结果

1
2
3
4
5
6
7
8
9
10
r.text: {"code":0,"msg":"success","obj":{"name":"小明","sex":"男","token":"psvmc"}}
r.json(): {'code': 0, 'msg': 'success', 'obj': {'name': '小明', 'sex': '男', 'token': 'psvmc'}}
r.content: b'{"code":0,"msg":"success","obj":{"name":"\xe5\xb0\x8f\xe6\x98\x8e","sex":"\xe7\x94\xb7","token":"psvmc"}}'
r.status_code: 200
r.encoding: utf_8
r.url: https://www.psvmc.cn/login.json?keyword=123
r.cookies: <Cookies[]>
r.headers: Headers({'server': 'nginx/1.14.0 (Ubuntu)', 'date': 'Fri, 26 Nov 2021 02:23:03 GMT', 'content-type': 'application/json', 'content-length': '78', 'last-modified': 'Thu, 25 Nov 2021 10:57:01 GMT', 'connection': 'keep-alive', 'etag': '"619f6bfd-4e"', 'accept-ranges': 'bytes'})
r.headers['Content-Type']: application/json
r.request.headers: Headers({'host': 'www.psvmc.cn', 'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx/0.21.1'})

判断返回状态码

1
r.status_code == httpx.codes.OK

POST请求

基本请求

1
2
r = httpx.post('https://www.psvmc.cn/login.json', data={'key':'value'})
r = httpx.post('https://www.psvmc.cn/login.json', json={'key':'value'})

文件上传

1
2
files ={'upload-file': open('report.xls','rb')}
r = httpx.post(url, files=files)

文件和数据

1
2
3
data = {'message': 'Hello, world!'}
files = {'file': open('report.xls', 'rb')}
r = httpx.post("https://httpbin.org/post", data=data, files=files)

二进制数据

1
2
content = b'Hello World'
response = httpx.post('http://127.0.0.1:5000/test/post', content=content)

其它请求

1
2
3
4
r = httpx.put('https://www.psvmc.cn/login.json', data={'key':'value'})
r = httpx.delete('https://www.psvmc.cn/login.json')
r = httpx.head('https://www.psvmc.cn/login.json')
r = httpx.options('https://www.psvmc.cn/login.json')

设置超时时间

1
2
3
4
import httpx

r = httpx.get('https://www.psvmc.cn/login.json', timeout=1)
print(r.text)

SSL

1
response = httpx.get('https://example.org', verify='../../client.pem')

又或者，你可以将verify设置为False禁用SSL验证：

1
response = httpx.get('https://example.org', verify=False)

自定义Header

1
2
headers ={'user-agent':'psvmc/0.0.1'}
r = httpx.get(url, headers=headers)

认证方式

HTTPX支持基本和摘要HTTP身份验证。

要提供基本身份验证凭据，请将2个元组的纯文本 str或 bytes对象作为 auth参数传递给请求函数：

1
2
3
4
5
6
7
import httpx

r = httpx.get(
    "https://www.psvmc.cn/login.json",
    auth=("my_user", "password123")
)
print(r.text)

要提供摘要式身份验证的凭据，您需要 DigestAuth使用纯文本用户名和密码作为参数实例化一个对象。然后可以将该对象作为 auth参数传递给上述请求方法：

1
2
3
4
5
import httpx

auth = httpx.DigestAuth("my_user", "password123")
r = httpx.get("https://www.psvmc.cn/login.json", auth=auth)
print(r.text)

异步请求

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


async def myrequest():
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            'https://www.psvmc.cn/login.json',
            params={'keyword': '123'}
        )
        result = resp.text
        print(result)
  

loop = asyncio.get_event_loop()
loop.run_until_complete(myrequest())

响应

常用的响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
print("r.text:", r.text)
print("r.json():", r.json())
print("r.content:", r.content)

# 响应码
print("r.status_code:", r.status_code)
# 响应的编码
print("r.encoding:", r.encoding)
# 请求的URL
print("r.url:", r.url)
# Cookie
print("r.cookies:", r.cookies)
# 响应的header
print("r.headers:", r.headers)
print("r.headers['Content-Type']:", r.headers['Content-Type'])
# 请求的header
print("r.request.headers:", r.request.headers)

流响应

对于大型下载，您可能需要使用不将整个响应主体立即加载到内存中的流式响应。

您可以流式传输响应的二进制内容…

1
2
3
4
5
import httpx

with httpx.stream("GET", "https://www.psvmc.cn/login.json") as r:
    for data in r.iter_bytes():
        print(data)

或回应文字

1
2
3
4
5
import httpx

with httpx.stream("GET", "https://www.psvmc.cn/login.json") as r:
    for text in r.iter_text():
        print(text)

或逐行流文本

1
2
3
4
5
import httpx

with httpx.stream("GET", "https://www.psvmc.cn/login.json") as r:
    for line in r.iter_lines():
        print(line)

HTTPX将使用通用行结尾，将所有情况标准化为 \n。

在某些情况下，您可能希望在不应用任何HTTP内容解码的情况下访问响应上的原始字节。在这种情况下的任何内容编码web服务器已诸如施加 gzip， deflate或 brotli将不会自动解码。

1
2
3
4
5
import httpx

with httpx.stream("GET", "https://www.psvmc.cn/login.json") as r:
    for chunk in r.iter_raw():
        print(chunk)

如果您以上述任何一种方式使用流式响应，则 response.contentand response.text属性将不可用，并且如果访问将引发错误。但是，您还可以使用响应流功能来有条件地加载响应主体：

1
2
3
4
5
6
import httpx

with httpx.stream("GET", "https://www.psvmc.cn/login.json") as r:
    if int(r.headers['Content-Length']) < 1000:
        r.read()
        print(r.text)

二进制加载为图片

1
2
3
from PIL import Image
from io import BytesIO
i =Image.open(BytesIO(r.content))