Python网络请求-httpx及URL拼接

前言

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

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

那么怎么选择呢

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

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

httpx的特点：

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

URL处理

获取URL中文件名

url = "node01/AAAAAA/31/marking/paper/2025/20250106/202501062001.png"
filename = url.rsplit('/', 1)[-1]
print(filename)

解释

rsplit('/', 1)[-1]: 使用rsplit从右边开始分割路径，只分割一次，然后取最后一个部分，即文件名称。

获取文件后缀

url = "node01/AAAAAA/31/marking/paper/2025/20250106/202501062001.png"
suffix = url.rsplit('.', 1)[-1]
print(suffix)

URL拼接

使用这种方式不用考虑路径中是/是否缺少或重复。

from urllib.parse import urljoin

base_url = "https://www.psvmc.cn"
path = "search"
url = urljoin(base_url, path)
print(url)

安装

pip install httpx

同步请求

GET请求

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)

结果

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'})

判断返回状态码

r.status_code == httpx.codes.OK

文件下载

import httpx

def download_file(url, filepath_all):
    with httpx.stream('GET', url) as r:
        try:
            r.raise_for_status()
            with open(filepath_all, 'wb') as f:
                for chunk in r.iter_bytes():
                    f.write(chunk)
        except Exception as e:
            print(f"download fail: {url}")

url = 'https://www.psvmc.cn/login.json'
filename = 'login.json'
download_file(url, filename)

POST请求

基本请求

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

文件上传

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

文件和数据

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

二进制数据

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

其它请求

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')

设置超时时间

import httpx

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

SSL

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

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

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

自定义Header

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

认证方式

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

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

import httpx

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

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

import httpx

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

异步请求

Get请求

import httpx
import asyncio


async def user_login():
    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(user_login())

Post请求

import httpx
import asyncio


async def user_login():
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            'https://www.psvmc.cn/login.json',
            data={'keyword': '123'}
        )
        result = resp.text
        return result
  

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

下载文件

async def download_file_async(url, filepath_all,download_begin_cb=None,download_progress_cb=None,download_end_cb=None):
    client = httpx.AsyncClient()
    async with client.stream('GET', url) as r:
        r.raise_for_status()
        # 获取响应头中的 Content-Length 字段，用于设置进度条的总大小
        total_size = int(r.headers.get("Content-Length", 0))
        download_size = 0
        download_begin_cb()
        with open(filepath_all, "wb") as f:
            async for chunk in r.aiter_bytes():
                f.write(chunk)
                download_size+=len(chunk)
                download_progress_cb(download_size,total_size)
        download_end_cb()

调用

def beginDownloadAppNew(self):
    appUrl = GlobalVars.appUrl

    def download_progress_cb(x, y):
        self.showUpdateProgress.emit(float(1.0 * x / y))

    def download_end_cb():
        self.showUpdateSuccess.emit()
        # 安装应用
        GlobalVars.mainSignalObj.installAppSignal.emit()

    def run_async_task():
        asyncio.run(
            download_file_async(
                appUrl,
                "app.exe",
                download_begin_cb=lambda: self.showUpdateProgress.emit(0),
                download_progress_cb=download_progress_cb,
                download_end_cb=download_end_cb,
            )
        )

    # 创建并启动新线程
    thread = threading.Thread(target=run_async_task)
    thread.start()

响应

常用的响应

# 返回str（字符串）
print("r.text:", r.text)

# 返回bytes（字节流）
print("r.content:", r.content)

# 返回json对象
print("r.json():", r.json())

其他

# 响应码
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)

流响应

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

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

import httpx

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

或回应文字

import httpx

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

或逐行流文本

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将不会自动解码。

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属性将不可用，并且如果访问将引发错误。但是，您还可以使用响应流功能来有条件地加载响应主体：

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)

二进制加载为图片

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