Python Requests - 重新认证后重试请求

Question

我正在使用Python 3.5.1 和Requests 2.9.1。我的用例如下：在向资源服务器 R 发出请求时，我需要从服务 T 进行身份验证（获取令牌）并将其用作 Authorization 标头的值。令牌在某个时候到期，需要一个新令牌被取走。

我有一个使用requests 的应用程序，它在启动时首先获取一个令牌并记住它 - 设置在用于对 R 的请求的Session 中。从那时起 3 分钟，一切都像一个魅力。 3 分钟后，我收到 unauthorized 响应，因为令牌无效。

我将Session 用于所有请求除了用于身份验证；此调用会更新 Session 上的 Authorization 标头以供其他请求使用。

我创建了代码以在检测到unauthorized 时自动重新验证，使用response 挂钩（仅在Session 上设置），代码如下：

def __hook(self, res, *args, **kwargs):
    if res.status_code == requests.codes.unauthorized:
        print('Token expired, refreshing')
        self.auth() # sets the token on self.__session

        req = res.request
        print('Resending request', req.method, req.url, req.headers)
        req.headers['Authorization'] = self.__session.headers['Authorization'] # why is it needed?

        return self.__session.send(res.request)

基本上，它甚至可以工作。不过有几个问题：

1. 为什么即使会话已更新并用于重新发送原始请求，也需要重新设置请求上的Authorization 标头？没有该行，应用程序将继续刷新令牌，因为新令牌从未使用过，这可以在输出中看到，令牌是导致自动刷新的原始令牌。

2. 如何使代码更健壮，即防止无限递归（我不确定在现实中是否可能，但在重试请求上设置Authorization 标头的行将继续一个开）？我正在考虑设置一个自定义标头，如果钩子发现失败的请求有它，它就不会重新验证并重新发送。有更好的吗？ 编辑： 事实证明，如果配置错误，可能会获得（几乎）无限循环（毕竟它是递归的）：令牌被用于一个环境（如 STAGING），但是资源服务器来自另一个 (TEST) - auth 请求将成功，但资源服务器的令牌实际上不正确。目前我实现了上面提到的“特殊”标头解决方案。

这是一般的好方法还是有什么更适合requests中的任务？

Answer 1

我想出的最终代码来解决这个问题。它比OP更完整。它公开了一个会话属性，您可以将其用于经过身份验证的请求。

之所以必须调用 auth，是因为 .send 只是发送 PreparedRequest。

如果您不喜欢我为防止无限递归而添加的 REATTEMPT 标头。您也可以像 OP 中提到的那样取消注册钩子（但您必须再次注册它以供将来使用），或者在我的编码类中使用 self._session 而不是 self.session 执行新请求。 （self._session 只是一个没有钩子的会话。）

import requests
from urllib.parse import urljoin

class JWTAuth:

    def __init__(self, user, passw, base):
        self.base = base
        self.user, self.passw = user, passw

        self._session = requests.Session()  # Session for tokens
        self.authenticate()

        self.session = requests.Session()  # Authenticated session
        self.session.auth = self.auth
        self.session.hooks['response'].append(self.reauth)

    def abs_url(self, path):
        """Combine the base url with a path."""
        return urljoin(self.base, path)

    def auth(self, req):
        """Just set the authentication token, on every request."""
        req.headers['Authorization'] = f'JWT {self.access}'
        return req

    def reauth(self, res, *args, **kwargs):
        """Hook to re-authenticate whenever authentication expires."""
        if res.status_code == requests.codes.unauthorized:
            if res.request.headers.get('REATTEMPT'):
                res.raise_for_status()
            self.refresh_auth()
            req = res.request
            req.headers['REATTEMPT'] = 1
            req = self.session.auth(req)
            res = self.session.send(req)
            return res

    def refresh_auth(self):
        """Use the refresh token to get a new access token."""
        res = self._session.post(self.abs_url("api/token/refresh/"), data={"refresh": self.refresh})
        if res.status_code == 200:
            self.access = res.json()['access']
        else:
            # Token expired -> re-authenticate
            self.authenticate()

    def authenticate(self):
        res = self._session.post(
            self.abs_url("api/token/"),
            data={"username": self.user, "password": self.passw},
        )
        res.raise_for_status()
        data = res.json()
        self.refresh, self.access = data['refresh'], data['access']

Answer 2

这是相同的解决方案，但没有使用类

session = requests.Session()
session.headers.update({"Authorization": f"Bearer deliberate-wrong-token"})

def refresh_token(r, *args, **kwargs):
    if r.status_code == 401:
        logger.info("Fetching new token as the previous token expired")
        token = get_token()
        session.headers.update({"Authorization": f"Bearer {token}"})
        r.request.headers["Authorization"] = session.headers["Authorization"]
        return session.send(r.request, verify=False)

session.hooks['response'].append(refresh_token)

Answer 3

所以我联系了 Requests 的作者，讨论了一个不相关的问题并询问了这个问题，结果证明这种方法是可以的。我所做的唯一更改是删除自定义标头（它确实有效并且还可以），现在我只是在重新发送请求之前取消注册挂钩中的响应挂钩。这打破了循环，一切都很好。