Python -- httpx库的基础应用
前言:
httpx
+ asyncio
是 Python 中用于 高效处理 HTTP 请求 的组合,尤其适合在 异步编程环境(如 FastAPI、爬虫、微服务)中使用。
相比于 requests 框架仅支持同步请求,所有操作都是阻塞式的。httpx 同时支持同步和异步请求(基于 asyncio
),适合与 FastAPI、Starlette、asyncio 等异步框架配合使用。
Python 实操:
1、安装 httpx
pip install httpx
如果想要支持 HTTP/2 协议,则需要额外安装 http2
支持:
pip install 'httpx[http2]'
2、get 请求
方式一:
httpx.get()
快捷方法。
适合写脚本或快速测试,但在正式项目中不推荐。
import httpx
import json
class httpxMethods:
@staticmethod
def getMethod():
# get请求
getUrl = "http://127.0.0.1:8080/getStudentInfo"
payload = {'id': '1'}
res2 = httpx.get(getUrl, params=payload).json()
json_str = json.dumps(res2, indent=4, ensure_ascii=False)
return json_str
if __name__ == "__main__":
method = httpxMethods()
print(method.getMethod())
与requests库不同,接口携带请求参数时,需要通过关键字参数 params=
来传递。
方式二(推荐):
使用 httpx.Client()
,并配合 with
上下文管理器。
with
语句可以确保连接在使用完后被正确关闭,避免资源泄漏。
相比于方式一,更规范、更高效、更安全地管理连接和资源。
import httpx
import json
class httpxMethods:
@staticmethod
def getMethod():
# GET 请求的 URL
getUrl = "http://127.0.0.1:8080/getStudentInfo"
payload = {'id': '1'}
# 使用 with 管理 Client 生命周期
with httpx.Client() as client:
response = client.get(getUrl, params=payload)
res_json = response.json()
# 格式化输出 JSON
json_str = json.dumps(res_json, indent=4, ensure_ascii=False)
return json_str
if __name__ == "__main__":
method = httpxMethods()
print(method.getMethod())
3、post 请求
方式一:
httpx.post() 快捷方式。
适合写脚本或快速测试,但在正式项目中不推荐。
import httpx
import json
class httpxMethods:
@staticmethod
def postMethod():
# post请求
postUrl = "http://127.0.0.1:8080/addStudentInfoTo1"
data = {
"id": 1000,
"classId": 10,
"name": "嘻哈",
"age": 11,
"sex": "男",
"idNumber": "445221199602104132",
"birthDay": "1996-01-10",
"createTime": "2025-05-06T17:35:00"
}
header = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
}
response = httpx.post(postUrl, json=data, headers=header).json()
json_str = json.dumps(response, indent=4, ensure_ascii=False)
return json_str
if __name__ == "__main__":
method = httpxMethods()
print(method.postMethod())
方式二(推荐):
使用 httpx.Client()
,并配合 with
上下文管理器。
import httpx
import json
class httpxMethods:
@staticmethod
def postMethod():
# POST 请求的 URL
postUrl = "http://127.0.0.1:8080/addStudentInfoTo1"
# 请求体数据
data = {
"id": 1000,
"classId": 10,
"name": "嘻哈",
"age": 11,
"sex": "男",
"idNumber": "445221199602104132",
"birthDay": "1996-01-10",
"createTime": "2025-05-06T17:35:00"
}
# 请求头
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
}
# 使用 with 管理 Client 生命周期
with httpx.Client() as client:
response = client.post(postUrl, json=data, headers=headers)
res_json = response.json()
# 格式化输出 JSON
json_str = json.dumps(res_json, indent=4, ensure_ascii=False)
return json_str
if __name__ == "__main__":
method = httpxMethods()
print(method.postMethod())
4、文件上传
import httpx
class httpxMethods:
@staticmethod
def fileUploadMethod(file_path):
# 文件上传
url = 'http://127.0.0.1:8080/api/files/upload'
try:
with open(file_path, 'rb') as f:
# 可自定义文件名上传,这里定义同名的test.log
files = {'file': ('test.log', f)}
response = httpx.post(url, files=files, timeout=10)
print(f"Status Code: {response.status_code}")
print("Response Text:", response.text)
except FileNotFoundError:
print("错误:文件未找到,请检查文件路径。")
except httpx.ConnectError:
print("错误:无法连接到目标服务器,请确认服务是否启动。")
except httpx.HTTPError as e:
print(f"请求过程中发生错误: {e}")
if __name__ == "__main__":
method = httpxMethods()
file_path = '../resources/test.log'
method.fileUploadMethod(file_path)
5、文件下载
import httpx
import os
@staticmethod
def fileDownloadMethod():
# 文件下载
url = 'http://127.0.0.1:8080/api/files/download/test.log'
file_path = '../resources/test4.log'
try:
with httpx.Client(timeout=10) as client:
with client.stream("GET", url, follow_redirects=True) as response:
response.raise_for_status() # 如果状态码不是 2xx,抛出异常
# 确保目标目录存在
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'wb') as f:
for chunk in response.iter_bytes(1024):
f.write(chunk)
print("下载完成")
except httpx.RequestError as e:
print(f"请求失败: {e}")
except Exception as e:
print(f"发生未知错误: {e}")
if __name__ == "__main__":
method = httpxMethods()
method.fileDownloadMethod()
-
with httpx.Client(...) as client:
创建一个客户端会话,自动管理连接资源释放; -
timeout=10
:设置最大等待时间为 10 秒; -
client.stream("GET", url, follow_redirects=True)
:以流式方式发起 GET 请求;stream=True
表示不一次性加载整个响应内容到内存中,适合大文件下载;follow_redirects=True
表示自动跟随重定向(如服务器返回 302 状态码时);
-
response.raise
如果响应状态码不是 2xx(比如 404、500),将抛出异常; -
打开文件并写入数据(流式)
'wb'
:表示以二进制写模式打开文件;response.iter_bytes(1024)
:每次读取 1024 字节(1KB);for chunk in ...
:循环读取并写入本地文件;- 流式下载避免了将整个文件加载到内存中,适合下载大文件。
补充:
java服务端代码:
https://gitee.com/a_cloud/python_springboot_demo.git
python客户端代码:
https://gitee.com/a_cloud/self_study_python.git
评论区