Python3: 同时支持 HTTP/2、同步、异步的 HTTP 客户端 HTTPX

- 日理万妓 2021-09-22 00:32 405阅读 0赞

本文链接: [https://xiets.blog.csdn.net/article/details/117638972][https_xiets.blog.csdn.net_article_details_117638972]

[Python3 学习笔记（目录）][Python3]

# 1. HTTPX #

`httpx` 模块是 Python 3 的全功能 HTTP 客户端，它提供了 同步 和 asyncio 异步 API，并支持 HTTP/1.1 和 **HTTP/2**。

`httpx` 是一个高性能的异步 HTTP 客户端，它建立在 `requests` 完善的可用性之上，支持 **连接和连接池的保持**、Cookie 持久性会话、自动内容解码、Basic/Digest 身份认证、HTTP(S) 代理、分段文件上传、分块请求 等一系列功能。`httpx` 的 API 布局大部分遵循了 `requests`，但比后者更强大，是用于 Python 的下一代 HTTP 客户端。

相关链接：

*  httpx 官网文档: [https://www.python-httpx.org/][https_www.python-httpx.org]
 *  httpx API 文档: [https://www.python-httpx.org/api/][https_www.python-httpx.org_api]
 *  GitHub 仓库: [https://github.com/encode/httpx/][https_github.com_encode_httpx]

HTTP 请求测试用地址: [http://httpbin.org/][http_httpbin.org] 或 [https://httpbin.org/][https_httpbin.org]

> HTTPX 需要 Python 3.6+

安装 `httpx` 模块：

$ pip3 install httpx

要包含 HTTP/2 的支持，，请使用：

$ pip3 install httpx[http2]

要包括可选的 brotli 解码器支持，请使用：

$ pip3 install httpx[brotli]

# 2. HTTPX 的请求方法 #

HTTP 请求方法：

httpx.get()                 # GET 请求
    httpx.post()                # POST 请求
    httpx.put()                 # PUT 请求
    httpx.patch()               # PATCH 请求
    httpx.head()                # HEAD 请求
    httpx.options()             # OPTIONS 请求
    httpx.delete()              # DELETE 请求

以上各请求最终都发到 `httpx.request()` 方法（参数列表也与之相似），并返回 `httpx.Response` 对象。

[httpx.request(method, url, …)][httpx.request_method_ url_] 方法的参数说明：

# HTTP 请求方法
    httpx.request(
        method: str,
        url: URLTypes,
        *,
        params: QueryParamTypes = None,
        content: RequestContent = None,
        data: RequestData = None,
        files: RequestFiles = None,
        json: typing.Any = None,
        headers: HeaderTypes = None,
        cookies: CookieTypes = None,
        auth: AuthTypes = None,
        proxies: ProxiesTypes = None,
        timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
        allow_redirects: bool = True,
        verify: VerifyTypes = True,
        cert: CertTypes = None,
        trust_env: bool = True,
    ) -> httpx.Response
    
    # 参数说明:
    # 除了 method 和 url 参数, 其他都是可选参数。
    #
    # method:
    # 请求方法: "GET", "POST", "PUT", "PATCH", "HEAD", "OPTIONS", or "DELETE"
    #
    # url:
    # 请求链接, 可以是 str 类型。
    #
    # params:
    # URL查询参数, 拼接到 URL Query String 中。
    #
    # 类型: 
    # str: "name1=value1&name2=value2"
    # dict: {name1: value1, name2: value2, ...}
    # list[tuple]: [(name1, value1), (name2, value2), ...]
    #
    # 拼接后结果如: http://httpbin.org/get?name1=value1&name2=value2
    #
    # content:
    # 要包含在请求 Body 中的二进制内容, 类型为 bytes 或 byte iterator。
    # byte iterator 可以是以 二进制打开的文件 IO 对象, 可用此参数直接上传文件。
    #
    # data:
    # 要包含在请求 Body 中的 form 表单数据, dict 类型。
    # 例如: data={name1: value1, name2: value2}
    # 将按 form 表单格式拼接后发送到 请求 Body 中, 如: "name1=value1&name2=value2"
    # 并自动添加请求头: "Content-Type": "application/x-www-form-urlencoded"
    #
    # files:
    # 要包含在请求 Body 中 form 表单文件字段, dict 类型。
    # 可结合 data 参数同时上传 form 表单的 普通字段 和 文件字段。
    #
    # 使用此参数, 自动添加请求头:
    # "Content-Type": "multipart/form-data; boundary=xxx"
    #
    # 参数格式:
    # {"field_name": file-like-objects}
    # {"field_name": ("filename", file-like-objects)}
    # {"field_name": ("filename", file-like-objects, "content_type")}
    #
    # 字段说明:
    # field_name: form 表单中文件字段的字段名
    # file-like-objects: 已打开的二进制文件 IO 流对象（上次完毕后需手动关闭流）
    # filename: 显式地设置 此文件字段的 文件名（默认使用本地文件名）
    # content_type: 显式地设置 此文件字段的 文件类型（默认根据文件名后缀设置）
    #
    # 其中 file-like-objects 可以用 str/bytes 来代替文件流的内容,
    # 可以同时上传多个文件。
    #
    # 参数传值示例:
    # {"f1": open("aa.jpg", "rb")} 提交后表单字段值为: name=f1, filename=aa.jpg
    # {"f1": open("aa.jpg", "rb"), "f2": open("bb.png", "rb")} 同时上传多个文件
    # {"f1": ("bb.jpg", open("aa.jpg", "rb"))} 提交后表单字段值为: name=f1, filename=bb.jpg
    #
    # json:
    # 要包含在请求 Body 中的 JSON 可序列化对象（如 dict 类型）,
    # 序列为为 JSON 字符串后发送到请求 Body 中。
    #
    # 使用此参数, 自动添加请求头:
    # "Content-Type": "application/json"
    #
    # 参数示例:
    # json={"name": "tom", "age": 30}
    #
    # headers:
    # 要包含在请求头中的 HTTP 请求头, 会覆盖现有的请求头, 一般为 dict 类型。
    #
    # 参数示例:
    # headers={"Header1": "Value1", "Header2": "Value2"}
    #
    # cookies:
    # 要包含在请求头中的 Cookies, dict 或 httpx.Cookies 或 http.cookiejar.CookieJar 类型,
    # 如果传递的是 CookieJar 对象, 则从中提取匹配此 URL 请求的 Cookie 添加到请求头中。
    #
    # 参数示例:
    # cookies={"name1": "value1", "name2": "value2"}
    #
    # auth:
    # 发送请求时使用的身份认证。
    # HTTP Auth 认证类型: Basic、Digest、Custom 等
    #
    # 参数示例:
    # Basic Auth: auth=httpx.BasicAuth("user", "pass")
    # 或 
    # auth=("user", "pass")
    # Digest Auth: auth=httpx.DigestAuth("user", "pass")
    #
    # 使用示例:
    # resp = httpx.get("https://httpbin.org/basic-auth/user/pass", auth=("user", "pass"))
    #
    # proxies:
    # 使用代理请求, str 或 dict 类型。
    # 
    # 参数示例:
    # # 所有请求发送到 http://localhost:8080 代理
    # proxies="http://localhost:8080"
    #
    # # 需要认证的代理连接格式
    # proxies="http://user:passwd@localhost:8080"
    #
    # # proxy_key 为 URL 前缀, 匹配此前缀的 URL 发送到 proxy_url 代理。
    # # proxy_key 值可以为 "http://"、"http://httpbin.org" 等。
    # # proxy_key 只包括 protocol://hostname[:port], 不包括 path,
    # # proxy_key 只匹配到 protocol://hostname[:port], 不匹配 path.
    # proxies={"proxy_key": "proxy_url"}
    #
    # proxies={ 
    # "http://": "http://localhost:8080", # http://* 格式的请求发送到指定代理
    # "https://": "http://localhost:8080" # https://* 格式的请求发送到指定代理
    # }
    #
    # timeout:
    # 超时时间, float 或 httpx.Timeout 类型。
    # 用 float 类型则表示所有操作的超时时间均为此时间, 默认均为 5.0 秒。
    # None 表示不检查超时。
    #
    # 可以创建 httpx.Timeout 对象更细致的控制超时时间:
    # Timeout(
    # timeout=None, # 其他为 None 的字段的默认超时时间
    *               *,
    # connect=None, # 连接超时时间
    # read=None, # 读取超时时间
    # write=None, # 写入超时时间
    # pool=None # 从连接池中获取连接的超时时间
    # )
    #
    # Timeout(None) # 所有操作不超时
    # Timeout(5.0) # 所有操作 5 秒超时
    # Timeout(None, connect=5.0) # 连接 5 秒超时, 其他操作不超时
    # Timeout(5.0, read=10.0) # 读取 10 秒超时, 其他操作 5 秒超时
    #
    # 超时将抛出对应的异常:
    # httpx.ConnectTimeout
    # httpx.ReadTimeout
    # httpx.WriteTimeout
    # httpx.PoolTimeout
    #
    # allow_redirects:
    # 启用 或 禁用 HTTP 重定向, bool 类型, 默认为 True 启用。
    #
    # verify:
    # 是否校验 SSL 证书, 用于 HTTPS 请求时验证所请求主机的身份, 有内置信任的根证书。
    # 类型 为 str, bool 或 ssl.SSLContext。
    # 默认为 True（校验）, False 表示不校验（不安全）。
    # 可以传递 str 类型指定 SSL 证书文件的路径,
    # 或传递 ssl.SSLContext 的实例。
    #
    # 没有被 httpx 信任的 SSL 证书, 可以手动指定信任, 示例:
    # verify=/my-ca-cert.pem 
    # 或
    # verify=/my-ca-cert.cer
    #
    # 一个 pem/cer 证书文件中可以有多个证书。
    #
    # cert:
    # 被请求的主机用来验证客户端的 SSL 证书。
    # 类型为:
    # str: SSL certificate file path
    # 两元组: (certificate file, key file)
    # 三元组: (certificate file, key file, password)
    #
    # trust_env:
    # 启用或禁用环境变量用于配置

`httpx.request()` 会把响应正文一次性读取到内存，如果下载大文件，需要使用 `httpx.stream()` [流式响应][Link 1] 请求：

# 流式响应请求, 返回一个响应迭代对象
    httpx.stream(
        method: str,
        url: URLTypes,
        *,
        params: QueryParamTypes = None,
        content: RequestContent = None,
        data: RequestData = None,
        files: RequestFiles = None,
        json: typing.Any = None,
        headers: HeaderTypes = None,
        cookies: CookieTypes = None,
        auth: AuthTypes = None,
        proxies: ProxiesTypes = None,
        timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
        allow_redirects: bool = True,
        verify: VerifyTypes = True,
        cert: CertTypes = None,
        trust_env: bool = True,
    ) -> typing.Iterator[Response]

## 2.1 GET 请求示例 ##

import httpx
    
    resp = httpx.get("https://httpbin.org/get")
    
    print(resp.status_code)                     # HTTP 状态码
    print(resp.headers)                         # 响应头
    print(resp.cookies)                         # 响应的 cookies
    print(resp.text)                            # str 类型的响应内容
    print(resp.content.decode("utf-8"))         # bytes 类型的响应内容

## 2.2 POST 请求示例 ##

import httpx
    
    resp = httpx.post("https://httpbin.org/post", content=b"Hello World")
    
    print(resp.status_code)
    print(resp.text)

## 2.3 提交 form 表单示例 ##

import httpx
    
    file1 = open("aa.txt", "rb")
    
    resp = httpx.post("https://httpbin.org/post",
                      data={ "key1": "value1"},
                      files={ "file_key1": file1}
                      )
    
    print(resp.content.decode("utf-8"))
    
    file1.close()

# 3. 响应对象: httpx.Response #

`httpx.request()` 发出请求后，返回一个响应对象 `httpx.Response`，包含了 响应状态、响应头、响应 Body 等各类信息。

[httpx.Response][] 类的属性和方法：

# 构造方法
    def __init__(...)
    
    
    # HTTP 状态码, 如: 200, 302, 404
    status_code: int
    
    
    # 原因短语, 如: OK, FOUND, NOT FOUND
    reason_phrase: str
    
    
    # HTTP 版本, 如: "HTTP/2"、"HTTP/1.1"
    http_version: str
    
    
    # 请求的 URL
    url: httpx.URL
    
    
    # 响应头
    # 获取响应头的值: headers.get(key), headers[key]
    # 遍历所有响应头: headers.keys(), headers.items()
    headers: httpx.Headers
    
    
    # 响应内容, bytes 类型。
    # 如果响应还没有 read(), 则引发 ResponseNotRead。
    # 请求封装为 Request 对象由 Client 发送请求时, 有个 stream 参数。
    # 如果 stream=False (默认) 则响应成功后自动 read() 响应内容。
    # 如果 stream=True, 则不会主动 read() 响应内容。
    # httpx.get(), httpx.post() 等直接请求的方法默认 stream=False。
    # 下载大文件时需要使用 Client(stream=True) 或 httpx.stream() 发送请求, 然后通过迭代响应逐块读取。
    content: bytes
    
    
    # 把 content 响应内容按 encoding 解码为文本
    text: str
    
    
    # 响应的内容编码
    encoding: str
    
    
    # 是否为重定向
    is_redirect: bool
    
    
    # 响应的 Cookie
    # 获取 Cookie: cookies.get(key), cookies[key]
    # 遍历 Cookie: cookies.keys(), cookies.items()
    cookies: httpx.Cookies
    
    
    # 该响应的请求对象, 所有请求会先封装成 Request 对象再由 Client 发送请求
    request: httpx.Request
    
    
    # 下一个请求。
    # 如果 allow_redirects=False (不自动处理重定向) 且响应为重定向, 则 next_request 为重定向的 Request。
    # 如果 allow_redirects=True (默认), 即自动处理重定向请求, 则该值为 None。
    next_request: httpx.Request
    
    
    # 响应的历史, 当请求被自动重定向时记录之前的响应。
    # 例如当前 Response (200) 是重定向后的响应, 
    # 则 history 为自动触发当前响应的前面的 Response (30X)
    history: list[httpx.Response]
    
    
    # 返回完成整个请求/响应周期所需的时间。
    # 从发送请求到接收到响应并读取完所有响应内容调用 close() 关闭响应释放连接后所需的时间。
    # elapsed.total_seconds() 可以获取准确的所需时间秒数。
    elapsed: datetime.timedelta
    
    
    # 根据 状态码(4XX/5XX) 引发 HTTPStatusError, 
    # 如果没有错误不做任何操作。
    # HTTP 状态码: https://httpstatuses.com/
    def raise_for_status() -> None
    
    
    # 将 content 解析为 JSON (dict) 返回,
    # 响应内容必须为 JSON 格式, 否则引发 JSONDecodeError
    def json() -> dict
    
    
    # 读取响应内容 (一次性读取全部), 如果是下载大文件需要通过迭代逐块读取。
    def read() -> bytes
    
    
    # 迭代读取响应内容 (未经 gzip/deflate/brotli 解压的原始内容, 不推荐使用)。
    # chunk_size 为每次迭代返回的最大块大小 (下同), 默认为 None 表示一次迭代完所有。
    def iter_raw(chunk_size: int = None) -> bytes iterator
    
    
    # 迭代读取响应内容 (经过 gzip/deflate/brotli 解压后的内容, 推荐使用), 
    # 内部实际是通过 iter_raw() 迭代后再解压。
    def iter_bytes(chunk_size: int = None) -> bytes iterator
    
    
    # 以文本方式迭代读取响应内容 (使用 encoding 解码), 
    # 内部实际是通过 iter_bytes() 迭代后再解码。
    def iter_text(chunk_size: int = None) -> text iterator
    
    
    # 以文本方式逐行迭代读取响应内容,
    # 内部实际是通过 iter_text() 迭代后再解析为行。
    def iter_lines() -> text iterator
    
    
    # 关闭响应 并 释放连接。
    # 不需要主动调用, 当响应内容被读取完毕后会自动调用。
    def close() -> None
    
    
    # 异步版本的 read()
    async def aread() -> bytes
    
    
    # 异步版本的 iter_raw()
    async def aiter_raw(chunk_size: int = None) -> bytes iterator
    
    
    # 异步版本的 iter_bytes()
    async def aiter_bytes(chunk_size: int = None) -> bytes iterator
    
    
    # 异步版本的 iter_text()
    async def aiter_text(chunk_size: int = None) -> text iterator
    
    
    # 异步版本的 iter_lines() 
    async def aiter_lines() -> text iterator
    
    
    # 异步版本的 close()
    # 不需要主动调用, 当响应内容被读取完毕后会自动调用。
    async def aclose() -> None

## 3.1 GET 请求示例： ##

import httpx
    
    resp = httpx.get("https://httpbin.org/cookies/set/hello/world", allow_redirects=False)
    
    print(resp.status_code)                     # 302
    print(resp.reason_phrase)                   # "FOUND"
    print(resp.http_version)                    # "HTTP/1.1"
    
    for header in resp.headers.items():         # 迭代响应头
        print(f"{ header[0]}: { header[1]}")
    
    for cookie in resp.cookies.items():         # 迭代 Cookie
        print(f"{ cookie[0]} = { cookie[1]}")     # "hello = world"
    
    print(resp.read().decode())                 # 输出响应内容

## 3.2 下载大文件: 流式响应 ##

对于大型下载，你可能希望使用不会一次将整个响应正文加载到内存中的 [流式响应][Link 1]。

下载文件示例：

import httpx
    
    client = httpx.Client()
    
    req = httpx.Request("GET", "https://example/aa.zip")
    
    resp = client.send(req, stream=True)
    
    if resp.status_code == 200:
        with open("aa.zip", "wb") as f:
            for buf in resp.iter_bytes(1024):
                f.write(buf)

或使用直接使用 `httpx.stream()` 流式响应下载：

import httpx
    
    with httpx.stream("GET", "https://example/aa.zip") as resp:
        with open("aa.zip", "wb") as f:
            for buf in resp.iter_bytes(1024):
                f.write(buf)

httpx.stream() 内部也是通过创建 httpx.Client() 然后 client.send(req, stream=True) 发出请求。

# 4. 请求对象: httpx.Request #

httpx 发出的所有请求最终都是先封装为 `httpx.Request` 再由 Client 发送请求。

[httpx.Request][] 类的属性和方法：

# 构造方法, 参数含义与 httpx.request() 方法中的参数含义相同。
    # 其中 stream 为发送到请求 Body 中的字节流, 
    # 如果传递了 stream 参数, 则会忽略 content, data, files, json 这几个参数, 并且需要自行构建相应的请求头。
    class httpx.Request(
            method: typing.Union[str, bytes],
            url: typing.Union["URL", str, RawURL],
            *,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
            content: RequestContent = None,
            data: RequestData = None,
            files: RequestFiles = None,
            json: typing.Any = None,
            stream: typing.Union[SyncByteStream, AsyncByteStream] = None,
        )

代码示例：

import httpx
    
    client = httpx.Client()
    
    req = httpx.Request("GET", "https://httpbin.org/get")
    
    resp = client.send(req)
    
    print(resp.text)

# 5. 保存/加载本地 Cookie #

响应的 Cookie 可以保存到本地文件，下次再从本地文件加载使用：

import httpx
    import http.cookiejar
    
    # 创建一个 CookieJar 对象, 关联本地文件
    cookiejar = http.cookiejar.MozillaCookieJar()
    # 加载本地文件中的 Cookie（文件必须存在）
    # cookiejar.load("cookies.txt")
    # 当前当做是新的会话, 可以清除掉旧的会话 Cookie
    # cookiejar.clear_session_cookies()
    
    # 创建 httpx 的 Cookies 对象（Cookie 将最终保存到 cookiejar 中）
    cookies = httpx.Cookies(cookiejar)
    
    # GET 请求, 自动携带上匹配 URL 的 Cookie
    resp = httpx.get("https://www.baidu.com/", cookies=cookies)
    print(resp.text)
    
    # 更新响应中的新 Cookie 到 cookies（最终被更新到 cookiejar）
    for his_resp in resp.history:
        cookies.update(his_resp.cookies)    # 防止是重定向后的响应, 前面如果有响应也要添加/更新对应的 Cookie
    cookies.update(resp.cookies)            # 添加/更新本次响应的 Cookie
    
    # 保存 cookiejar 中的 Cookies 到本地文件
    cookiejar.save("cookies.txt")

# 6. 同步 HTTP 客户端: httpx.Client #

使用 httpx.get(), httpx.request() 等直接发送 HTTP 请求的方式是先内部封装好请求对象，然后再创建一个 HTTP 客户端对象 (httpx.Client)，然后由该客户端发送请求，响应完毕后立即关闭客户端。httpx.request() 每发一个请求都需要频繁地重新建立 TCP 连接、发送 HTTP 请求、接收响应、关闭 TCP 链接，而且还不能自动保持 Cookies，对于相同域名的 URL 请求则会显得非常低效。因此如果需要频繁发送请求，需要先创建一个 HTTP 客户端，然后统一由此客户端发送请求。

`httpx.Client` 是一个 HTTP 客户端，具有 **连接池保持连接**、**HTTP/2**、**Cookie 持久性** 等功能。

> HTTP/2 最大优势之一是支持相同域名的 URL 请求复用同一个 TCP 连接（前提是使用同一个连接池），不需要每个请求都频繁建立 TCP 连接，效率明显提高。目前大多数主流网站都已支持 HTTP/2。HTTP/2 介绍：[https://http2.github.io/][https_http2.github.io]。

## 6.1 httpx.Client 的构造方法 ##

[httpx.Client][] 类的构造方法：

class httpx.Client(
            *,
            auth: AuthTypes = None,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
            verify: VerifyTypes = True,
            cert: CertTypes = None,
            http2: bool = False,
            proxies: ProxiesTypes = None,
            mounts: typing.Mapping[str, BaseTransport] = None,
            timeout: TimeoutTypes = Timeout(timeout=5.0),
            limits: Limits = Limits(max_connections=100, 
                                    max_keepalive_connections=20, 
                                    keepalive_expiry=5.0),
            max_redirects: int = 20,
            event_hooks: typing.Mapping[str, typing.List[typing.Callable]] = None,
            base_url: URLTypes = "",
            transport: BaseTransport = None,
            app: typing.Callable = None,
            trust_env: bool = True,
        )
    
    # 参数说明:
    # 部分参数的用法与 httpx.request() 方法中的参数用法相同, 不再细说。
    # httpx.Client 中的参数最终会 全局应用（合并追加）到每一个 HTTP 请求,
    # 作为每一个 HTTP 请求的默认参数（相同参数会被具体请求时传递的参数覆盖）。
    #
    # auth: 
    # 发送请求时使用的身份验证
    #
    # params:
    # 要包含在请求 URL 中的查询参数, 作为字符串、字典或二元组序列。
    #
    # headers:
    # 发送请求时要包含的 HTTP 请求头字典。
    #
    # cookies:
    # 发送请求时要包含的 Cookie, dict 或 httpx.Cookies 或 http.cookiejar.CookieJar 类型。
    # 如果传递的是 dict, 则 Cookie 会添加到每一个 URL 请求。
    # 如果传递的是 CookieJar 对象, 则从中提取匹配 URL 请求的 Cookie 添加到请求头中。
    #
    # verify:
    # 是否校验 SSL 证书, 默认校验。
    #
    # cert:
    # 被请求的主机用来验证客户端的 SSL 证书。
    #
    # http2:
    # 是否开启 HTTP/2, 需要开启后才能使用 HTTP/2
    #
    # proxies:
    # 使用代理请求
    #
    # mounts:
    # mounts/挂载
    #
    # timeout:
    # 超时时间, 默认均为 5.0 秒
    #
    # limits: Limits = Limits(max_connections=100, 
    # 要使用的限制配置, httpx.Limits 类型。 
    # 默认为:
    # Limits(
    # max_connections=100, # 最大并发连接数
    # max_keepalive_connections=20, # 连接池最大保持的连接数
    # keepalive_expiry=5.0 # 保持的连接的过期时间, 默认 5 秒
    # )
    #
    # max_redirects:
    # 最大重定向响应数, 默认为 20
    #
    # event_hooks:
    # 事件钩子
    # 
    # base_url:
    # 基础 URL, 发送请求的 URL 如果是相对路径, 则拼接到 base_url 后面形成绝对路径再发请求。
    #
    #
    # transport:
    # 用于通过网络发送请求的传输类
    #
    # app:
    # 一个用于发送请求的 WSGI 应用程序，而不是发送实际的网络请求。
    #
    # trust_env
    # 启用或禁用环境变量用于配置

## 6.2 httpx.Client 的请求方法 ##

httpx.Client 对象的 HTTP 请求方法：

client.get()                    # GET 请求
    client.post()                   # POST 请求
    client.put()                    # PUT 请求
    client.patch()                  # PATCH 请求
    client.head()                   # HEAD 请求
    client.options()                # OPTIONS 请求
    client.delete()                 # DELETE 请求

以上各请求最终都发到 `client.request()` 方法（参数列表也与之相似），并返回 `httpx.Response` 对象。

`httpx.Client.request()` 方法的参数：

# 方法参数的使用与 httpx.request() 相同
    httpx.Client.request(
            method: str,
            url: URLTypes,
            *,
            content: RequestContent = None,
            data: RequestData = None,
            files: RequestFiles = None,
            json: typing.Any = None,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
            auth: typing.Union[AuthTypes, UnsetType] = UNSET,
            allow_redirects: bool = True,
            timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
        ) -> httpx.Response:

httpx.Client 的其他方法：

# 流式响应请求
    httpx.Client.stream(
            method: str,
            url: URLTypes,
            *,
            content: RequestContent = None,
            data: RequestData = None,
            files: RequestFiles = None,
            json: typing.Any = None,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
            auth: typing.Union[AuthTypes, UnsetType] = UNSET,
            allow_redirects: bool = True,
            timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
        ) -> typing.Iterator[Response]
    
    
    # 构造一个请求对象
    httpx.Client.build_request(
            method: str,
            url: URLTypes,
            *,
            content: RequestContent = None,
            data: RequestData = None,
            files: RequestFiles = None,
            json: typing.Any = None,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
        ) -> httpx.Request
    
    
    # 发出一个请求
    httpx.Client.send(
            request: Request,
            *,
            stream: bool = False,
            auth: typing.Union[AuthTypes, UnsetType] = UNSET,
            allow_redirects: bool = True,
            timeout: typing.Union[TimeoutTypes, UnsetType] = UNSET,
        ) -> httpx.Response
    
    
    # 关闭客户端（关闭传输和代理）
    httpx.Client.close() -> None

httpx.Client 支持 `with` 语句：

with httpx.Client() as client:
        resp = client.get("https://httpbin.org/get")
        print(resp.text)

## 6.3 httpx.Client 请求示例 ##

>>> import httpx
    >>> 
    >>> # 创建一个 HTTP 客户端
    >>> client = httpx.Client(
    ...     headers={ "User-Agent": "hello/world"},
    ...     http2=True
    ... )
    >>> 
    >>> # GET 请求
    >>> resp = client.get("https://httpbin.org/get?aa=bb&cc=123",
    ...                   params={ "dd": True, "xyz": "hello"})
    >>> 
    >>> print(resp.text)
    { 
      "args": { 
        "aa": "bb", 
        "cc": "123", 
        "dd": "true", 
        "xyz": "hello"
      }, 
      "headers": { 
        "Accept": "*/*", 
        "Accept-Encoding": "gzip, deflate", 
        "Host": "httpbin.org", 
        "User-Agent": "hello/world"
      }, 
      "origin": "127.0.0.1", 
      "url": "https://httpbin.org/get?aa=bb&cc=123&dd=true&xyz=hello"
    }
    >>>
    >>> # POST 请求
    >>> resp = client.post("https://httpbin.org/post", content="Hello World")
    >>> 
    >>> print(resp.text)
    { 
      "args": { }, 
      "data": "Hello World", 
      "files": { }, 
      "form": { }, 
      "headers": { 
        "Accept": "*/*", 
        "Accept-Encoding": "gzip, deflate", 
        "Content-Length": "11", 
        "Host": "httpbin.org", 
        "User-Agent": "hello/world"
      }, 
      "json": null, 
      "origin": "127.0.0.1", 
      "url": "https://httpbin.org/post"
    }
    >>>
    >>> # 提交 form 表单
    >>> with open("aa.txt", "rb") as file1:
    ...     resp = client.post("https://httpbin.org/post",
    ...                       data={ "key1": "value1"},
    ...                       files={ "file_key1": file1}
    ...                       )
    ...     print(resp.text)
    ... 
    { 
      "args": { }, 
      "data": "", 
      "files": { 
        "file_key1": "Text Content Demo"
      }, 
      "form": { 
        "key1": "value1"
      }, 
      "headers": { 
        "Accept": "*/*", 
        "Accept-Encoding": "gzip, deflate", 
        "Content-Length": "281", 
        "Content-Type": "multipart/form-data; boundary=xxxxxx", 
        "Host": "httpbin.org", 
        "User-Agent": "hello/world", 
      }, 
      "json": null, 
      "origin": "127.0.0.1", 
      "url": "https://httpbin.org/post"
    }
    >>>
    >>> # 关闭客户端
    >>> client.close()

# 7. 异步 HTTP 客户端: httpx.AsyncClient #

`httpx.request()` 和 `httpx.Client` 相关方法都是同步请求方法，httpx 还支持 asyncio 异步请求，httpx 的异步请求使用 `httpx.AsyncClient` 异步 HTTP 客户端。

`httpx.AsyncClient` 是 `httpx.Client` 的异步版本，与之有相同的功能，同样支持 **连接池保持连接**、**HTTP/2**、**Cookie 持久性** 等功能，相关方法需使用 async/await 语法调用。

`httpx.AsyncClient` 类的构造方法：

# 方法参数与 httpx.Client 相同
    class httpx.AsyncClient(
            *,
            auth: AuthTypes = None,
            params: QueryParamTypes = None,
            headers: HeaderTypes = None,
            cookies: CookieTypes = None,
            verify: VerifyTypes = True,
            cert: CertTypes = None,
            http2: bool = False,
            proxies: ProxiesTypes = None,
            mounts: typing.Mapping[str, AsyncBaseTransport] = None,
            timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
            limits: Limits = DEFAULT_LIMITS,
            max_redirects: int = DEFAULT_MAX_REDIRECTS,
            event_hooks: typing.Mapping[str, typing.List[typing.Callable]] = None,
            base_url: URLTypes = "",
            transport: AsyncBaseTransport = None,
            app: typing.Callable = None,
            trust_env: bool = True,
        )

httpx.AsyncClient 的方法和 httpx.Client 基本相同，不过前者的请求等相关会阻塞 IO 的方法为 async 异步方法：

class httpx.AsyncClient:
        async def get(...) -> Response          # GET 请求
        async def post(...) -> Response         # POST 请求
        async def put(...) -> Response          # PUT 请求
        async def patch(...) -> Response        # PATCH 请求
        async def head(...) -> Response         # HEAD 请求
        async def options(...) -> Response      # OPTIONS 请求
        async def delete(...) -> Response       # DELETE 请求
        
        async def request(...) -> Response      # 指定请求方法请求
        
        async def stream(...) -> AsyncIterator[Response]    # 流式请求
        
        def build_request(...) -> Request       # 构建请求（非异步方法）
        
        async def send(...) -> Request          # 发送请求
        
        async def aclose() -> None              # 关闭客户端
        
        async def __aenter__(...) -> self       # 异步上下文 进入方法
        async def __aexit__(...) -> None        # 异步上下文 退出方法

通过 httpx.AsyncClient 异步客户端发送请求返回的 Response 响应，需使用相关的 async 异步方法读取内容。

httpx.AsyncClient 代码示例：

import httpx
    import asyncio
    
    async def main():
        client = httpx.AsyncClient()
        resp = await client.get("https://httpbin.org/get")
        print(resp.text)        # 响应内容在 get() 返回响应之前已经是被(异步)读取完毕
        await client.aclose()   # 使用完毕后必须关闭客户端
    
    asyncio.run(main())

httpx.AsyncClient 支持 `ascyn with` 语句：

import httpx
    import asyncio
    
    async def main():
        async with httpx.AsyncClient() as client:
            resp = await client.get("https://httpbin.org/get")
            print(resp.text)
    
    asyncio.run(main())

httpx.AsyncClient 异步 GET、POST、流式请求 代码示例：

import httpx
    import asyncio
    
    
    async def main():
        client = httpx.AsyncClient()
        async with client:
            # GET 请求
            resp = await client.get("https://httpbin.org/get")
            print(resp.text)
            # 如果要迭代, 需使用 a 开头的异步方法异步迭代
            async for line in resp.aiter_lines():
                print(line)
    
            # POST 请求
            resp = await client.post("https://httpbin.org/post", content=b"Hello World")
            print(resp.text)
    
            # 提交 form 表单
            with open("data.txt") as file1:
                resp = await client.post("https://httpbin.org/post",
                                         data={ "key1": "value1"},
                                         files={ "file_key1": file1}
                                         )
                print(resp.text)
    
            # 流式请求 (一)
            async with client.stream("GET", "https://httpbin.org/get") as resp:
                # 需要使用 a 开头的方法异步迭代
                async for bs in resp.aiter_bytes(1024):
                    print(bs)
    
            # 流式请求 (二)
            req = client.build_request("GET", "https://httpbin.org/get")
            resp = await client.send(req, stream=True)
            # 需要使用 a 开头的方法异步迭代
            async for line in resp.aiter_lines():
                print(line)
    
    
    asyncio.run(main())

[https_xiets.blog.csdn.net_article_details_117638972]: https://xiets.blog.csdn.net/article/details/117638972
[Python3]: https://xiets.blog.csdn.net/article/details/109266600
[https_www.python-httpx.org]: https://www.python-httpx.org/
[https_www.python-httpx.org_api]: https://www.python-httpx.org/api/
[https_github.com_encode_httpx]: https://github.com/encode/httpx/
[http_httpbin.org]: http://httpbin.org/
[https_httpbin.org]: https://httpbin.org/
[httpx.request_method_ url_]: https://www.python-httpx.org/api/#helper-functions
[Link 1]: https://www.python-httpx.org/quickstart/#streaming-responses
[httpx.Response]: https://www.python-httpx.org/api/#response
[httpx.Request]: https://www.python-httpx.org/api/#request
[https_http2.github.io]: https://http2.github.io/
[httpx.Client]: https://www.python-httpx.org/api/#client