httpx 自定義身份驗證

發(fā)出請求或?qū)嵗?Client?時，?auth?參數(shù)可用于傳遞要使用的身份驗證方案。?auth?參數(shù)可能是以下之一...

• ?username?/?password?的雙元組，用于基本身份驗證。
• ?httpx.BasicAuth()?或?httpx.DigestAuth()? 的實例。
• 可調(diào)用的，接受請求并返回經(jīng)過身份驗證的請求實例。
• ?httpx.Auth?的子類的實例。

其中涉及最多的是最后一個，它允許您創(chuàng)建涉及一個或多個請求的身份驗證流。?httpx.Auth?的子類應(yīng)該實現(xiàn) ?def auth_flow(request)? ，并生成需要發(fā)出的任何請求...

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

    def auth_flow(self, request):
        # Send the request, with a custom `X-Authentication` header.
        request.headers['X-Authentication'] = self.token
        yield request

如果身份驗證流需要多個請求，您可以發(fā)出多個?yield?，并在每種情況下獲取響應(yīng)...

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

    def auth_flow(self, request):
      response = yield request
      if response.status_code == 401:
          # If the server issues a 401 response then resend the request,
          # with a custom `X-Authentication` header.
          request.headers['X-Authentication'] = self.token
          yield request

自定義身份驗證類設(shè)計為不執(zhí)行任何 I/O，以便它們可以同時用于同步和異步?Client?實例。如果要實現(xiàn)需要請求正文的身份驗證方案，則需要使用?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:
          # If the server issues a 401 response then resend the request,
          # with a custom `X-Authentication` header.
          request.headers['X-Authentication'] = self.sign_request(...)
          yield request

    def sign_request(self, request):
        # Create a request signature, based on `request.method`, `request.url`,
        # `request.headers`, and `request.content`.
        ...

同樣，如果要實現(xiàn)需要訪問響應(yīng)正文的方案，請使用?requires_response_body?屬性。然后，您將能夠訪問響應(yīng)正文屬性和方法，例如?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:
            # If the server issues a 401 response, then issue a request to
            # refresh tokens, and resend the request.
            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):
        # Return an `httpx.Request` for refreshing tokens.
        ...

    def update_tokens(self, response):
        # Update the `.access_token` and `.refresh_token` tokens
        # based on a refresh response.
        data = response.json()
        ...

如果確實需要執(zhí)行 HTTP 請求以外的 I/O（如訪問基于磁盤的緩存），或者需要使用并發(fā)基元（如鎖），則應(yīng)覆蓋 ?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

如果您只想支持這兩種方法之一，則仍應(yīng)重寫它，但應(yīng)提出顯式 ?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")