侧边栏壁纸
博主头像
一朵云的博客博主等级

拥抱生活,向阳而生。

  • 累计撰写 108 篇文章
  • 累计创建 28 个标签
  • 累计收到 7 条评论

目 录CONTENT

文章目录

Python -- httpx库的基础应用

一朵云
2023-04-21 / 0 评论 / 0 点赞 / 1331 阅读 / 8979 字

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

image-llga.png

与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())

image-jibp.png

方式二(推荐):

使用 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)

image-mosh.png

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

0

评论区