如何让Python的Requests/HTTPX为所有重定向URL添加认证？

针对httpx的解决方案 #

方法1：使用Client实例全局配置认证 #

将认证信息配置在httpx.Client中，而非单个请求里。这样客户端发起的所有请求（包括重定向触发的请求）都会自动携带认证信息：

import httpx

api_url = 'https://demo.vizionapi.com/carriers'

headers = {
  'X-API-Key':'API KEY'
}

# 创建带全局配置的客户端
with httpx.Client(
    verify='supporting_files/cacert.pem',
    auth=('my username', 'my password'),
    follow_redirects=True
) as client:
    response = client.get(api_url, headers=headers)

# 处理响应
print(response.status_code)

方法2：自定义请求钩子强制添加认证 #

如果需要更灵活的控制（比如仅对特定域名添加认证），可以通过event_hooks给每个请求（包括重定向请求）手动注入认证：

import httpx
from httpx._auth import BasicAuth

def inject_auth(request):
    # 为所有请求添加Basic认证
    auth = BasicAuth('my username', 'my password')
    auth.sync_auth_flow(request)

api_url = 'https://demo.vizionapi.com/carriers'

headers = {
  'X-API-Key':'API KEY'
}

client = httpx.Client(
    verify='supporting_files/cacert.pem',
    follow_redirects=True,
    event_hooks={'request': [inject_auth]}
)

response = client.get(api_url, headers=headers)
client.close()

针对requests的解决方案 #

requests默认不会将认证信息自动传递到跨域名的重定向请求中，需要通过以下方式处理：

方法1：手动处理重定向逻辑 #

关闭自动重定向，手动跟进每个重定向URL并携带认证：

import requests

api_url = 'https://demo.vizionapi.com/carriers'

headers = {
    'X-API-Key': 'API KEY'
}
auth = ('my username', 'my password')
verify = 'supporting_files/cacert.pem'

session = requests.Session()
session.auth = auth
session.verify = verify

# 发起初始请求，关闭自动重定向
response = session.get(api_url, headers=headers, allow_redirects=False)

# 手动跟进所有重定向
while response.status_code in (301, 302, 307, 308):
    redirect_url = response.headers['Location']
    response = session.get(redirect_url, headers=headers, allow_redirects=False)

# 处理最终响应
print(response.status_code)

方法2：自定义重定向处理器 #

重写默认的重定向处理器，强制在每个重定向请求中添加认证：

import requests
from requests.models import PreparedRequest
from requests.auth import HTTPBasicAuth
from requests.handlers import HTTPRedirectHandler

class AuthRedirectHandler(HTTPRedirectHandler):
    def __init__(self, auth):
        self.auth = auth
        super().__init__()

    def redirect_request(self, req, fp, code, msg, headers, new_url):
        # 构建新的重定向请求并强制添加认证
        new_req = PreparedRequest()
        new_req.prepare(
            method=req.method,
            url=new_url,
            headers=req.headers,
            files=req.files,
            data=req.data,
            params=req.params,
            auth=self.auth,
            cookies=req.cookies
        )
        return new_req

api_url = 'https://demo.vizionapi.com/carriers'

headers = {
    'X-API-Key': 'API KEY'
}
auth = HTTPBasicAuth('my username', 'my password')
verify = 'supporting_files/cacert.pem'

session = requests.Session()
session.verify = verify
# 替换默认的重定向处理器
session.redirect_handler = AuthRedirectHandler(auth)

response = session.get(api_url, headers=headers)

问题根源 #

你遇到的401错误，是因为默认情况下httpx/requests仅会将auth参数应用在初始请求上。当请求被公司代理重定向到新的URL时，认证信息不会自动传递过去，导致重定向请求未通过身份验证。

内容的提问来源于stack exchange，提问作者Vae Jiang