认证

认证信息既可以针对单个请求进行设置...

>>> auth = httpx.BasicAuth(username="username", password="secret")
>>> client = httpx.Client()
>>> response = client.get("https://www.example.com/", auth=auth)

也可以配置在客户端实例上，确保所有发出的请求都会携带认证凭据...

>>> auth = httpx.BasicAuth(username="username", password="secret")
>>> client = httpx.Client(auth=auth)
>>> response = client.get("https://www.example.com/")

基本认证(Basic Authentication)

HTTP基本认证是一种未加密的认证方案，它使用简单的用户名和密码编码方式置于请求的Authorization头部。由于未加密，通常只应在https连接上使用，尽管这不是强制要求。

>>> auth = httpx.BasicAuth(username="finley", password="secret")
>>> client = httpx.Client(auth=auth)
>>> response = client.get("https://httpbin.org/basic-auth/finley/secret")
>>> response
<Response [200 OK]>

摘要认证(Digest Authentication)

HTTP摘要认证是一种挑战-响应式的认证方案。与基本认证不同，它提供了加密机制，可以在未加密的http连接上使用。该方案需要额外的通信回合来完成认证协商。

>>> auth = httpx.DigestAuth(username="olivia", password="secret")
>>> client = httpx.Client(auth=auth)
>>> response = client.get("https://httpbin.org/digest-auth/auth/olivia/secret")
>>> response
<Response [200 OK]>
>>> response.history
[<Response [401 UNAUTHORIZED]>]

NetRC 认证

HTTPX 可以配置使用 .netrc 配置文件进行认证。

.netrc 配置文件允许将认证凭据与指定主机关联。当向 netrc 文件中存在的主机发起请求时，用户名和密码将通过 HTTP 基本认证方式包含在请求中。

示例 .netrc 文件：

machine example.org
login example-username
password example-password

machine python-httpx.org
login other-username
password other-password

以下是一些使用 httpx 配置 .netrc 认证的示例：

使用用户主目录下的默认 .netrc 文件：

>>> auth = httpx.NetRCAuth()
>>> client = httpx.Client(auth=auth)

使用显式指定的 .netrc 文件路径：

>>> auth = httpx.NetRCAuth(file="/path/to/.netrc")
>>> client = httpx.Client(auth=auth)

使用 NETRC 环境变量配置 .netrc 文件路径，若未设置则回退到默认路径：

>>> auth = httpx.NetRCAuth(file=os.environ.get("NETRC"))
>>> client = httpx.Client(auth=auth)

NetRCAuth() 类使用了 Python 标准库中的 netrc.netrc() 函数。如果 .netrc 文件未找到或无法解析，可能会抛出异常，更多细节请参阅相关文档。

自定义认证方案

在发起请求或实例化客户端时，可以使用 auth 参数传递要使用的认证方案。auth 参数可以是以下类型之一...

包含 username/password 的二元组，用于基本认证
httpx.BasicAuth()、httpx.DigestAuth() 或 httpx.NetRCAuth() 的实例
一个可调用对象，接收请求并返回认证后的请求实例
httpx.Auth 子类的实例

其中最复杂的是最后一种，它允许你创建涉及一个或多个请求的认证流程。httpx.Auth 的子类应实现 def auth_flow(request) 方法，并通过 yield 发出需要执行的所有请求...

class MyCustomAuth(httpx.Auth):
    def __init__(self, token):
        self.token = token

    def auth_flow(self, request):
        # 发送带有自定义 `X-Authentication` 请求头的请求
        request.headers['X-Authentication'] = self.token
        yield request

如果认证流程需要多个请求，你可以发出多个 yield，并在每种情况下获取响应...

class MyCustomAuth(httpx.Auth):
    def __init__(self, token):
        self.token = token

    def auth_flow(self, request):
      response = yield request
      if response.status_code == 401:
          # 如果服务器返回 401 响应，则重新发送请求
          # 并添加自定义的 `X-Authentication` 请求头
          request.headers['X-Authentication'] = self.token
          yield request

自定义认证类的设计原则是不执行任何 I/O 操作，因此它们可以同时用于同步和异步客户端实例。如果你实现的认证方案需要访问请求体，则需要在类上使用 requires_request_body 属性进行声明。

之后你就可以在 .auth_flow() 方法中访问 request.content。

class MyCustomAuth(httpx.Auth):
    requires_request_body = True

    def __init__(self, token):
        self.token = token

    def auth_flow(self, request):
      response = yield request
      if response.status_code == 401:
          # 如果服务器返回 401 响应，则重新发送请求
          # 并添加自定义的 `X-Authentication` 请求头
          request.headers['X-Authentication'] = self.sign_request(...)
          yield request

    def sign_request(self, request):
        # 基于 `request.method`、`request.url`、
        # `request.headers` 和 `request.content` 创建请求签名
        ...

类似地，如果你实现的方案需要访问响应体，则使用 requires_response_body 属性。之后你就可以访问响应体属性和方法，如 response.content、response.text、response.json() 等。

class MyCustomAuth(httpx.Auth):
    requires_response_body = True

    def __init__(self, access_token, refresh_token, refresh_url):
        self.access_token = access_token
        self.refresh_token = refresh_token
        self.refresh_url = refresh_url

    def auth_flow(self, request):
        request.headers["X-Authentication"] = self.access_token
        response = yield request

        if response.status_code == 401:
            # 如果服务器返回 401 响应，则发起刷新令牌的请求
            # 并重新发送原始请求
            refresh_response = yield self.build_refresh_request()
            self.update_tokens(refresh_response)

            request.headers["X-Authentication"] = self.access_token
            yield request

    def build_refresh_request(self):
        # 返回用于刷新令牌的 `httpx.Request`
        ...

    def update_tokens(self, response):
        # 基于刷新响应更新 `.access_token` 和 `.refresh_token` 令牌
        data = response.json()
        ...

如果你确实需要执行 HTTP 请求之外的 I/O 操作（例如访问基于磁盘的缓存），或者需要使用并发原语（如锁），那么你应该重写 .sync_auth_flow() 和 .async_auth_flow() 方法（而不是 .auth_flow()）。前者将被 httpx.Client 使用，后者将被 httpx.AsyncClient 使用。

import asyncio
import threading
import httpx


class MyCustomAuth(httpx.Auth):
    def __init__(self):
        self._sync_lock = threading.RLock()
        self._async_lock = asyncio.Lock()

    def sync_get_token(self):
        with self._sync_lock:
            ...

    def sync_auth_flow(self, request):
        token = self.sync_get_token()
        request.headers["Authorization"] = f"Token {token}"
        yield request

    async def async_get_token(self):
        async with self._async_lock:
            ...

    async def async_auth_flow(self, request):
        token = await self.async_get_token()
        request.headers["Authorization"] = f"Token {token}"
        yield request

如果你只想支持其中一种方法，仍然应该重写它，但要显式抛出 RuntimeError。

import httpx
import sync_only_library


class MyCustomAuth(httpx.Auth):
    def sync_auth_flow(self, request):
        token = sync_only_library.get_token(...)
        request.headers["Authorization"] = f"Token {token}"
        yield request

    async def async_auth_flow(self, request):
        raise RuntimeError("Cannot use a sync authentication class with httpx.AsyncClient")