Django中如何实现带基础认证的第三方API持久化Session复用
实现方案
你遇到的服务账号频繁被锁,核心原因就是每次请求新建requests.Session时,都会发起独立的身份认证流程,短时间大量重复认证请求会触发第三方API的暴力破解风控规则,直接锁定账号。复用持久化的Session实例不仅能解决这个问题,还能复用底层TCP连接,降低请求开销。
实现思路
利用Django的AppConfig.ready()钩子,这个方法会在Django应用完成启动加载、正式开始处理请求前执行且仅执行一次,在这个时机初始化携带认证信息的全局Session实例,所有请求直接复用该实例即可。
requests.Session本身是线程安全的,多线程部署场景下可以直接使用;如果是多进程部署(gunicorn/uWSGI多worker模式),每个worker进程会独立初始化自己的Session实例,不会出现跨进程冲突问题。
具体实现步骤
1. 抽离配置,避免硬编码凭证
首先把第三方API的凭证、地址从业务代码里抽离,放到Django配置中,通过环境变量读取,避免凭证泄露:
# 项目settings.py import os # 从环境变量读取第三方API配置 EXTERNAL_API_USERNAME = os.getenv("EXTERNAL_API_USERNAME") EXTERNAL_API_PASSWORD = os.getenv("EXTERNAL_API_PASSWORD") EXTERNAL_API_BASE_URL = "https://external.api.com"
2. 编写全局Session初始化逻辑
新建独立的工具模块管理第三方API会话,避免循环导入问题:
# utils/external_api_client.py import requests from django.conf import settings # 全局会话实例占位 external_api_session = None def init_external_session() -> None: """初始化持久化的第三方API会话,应用启动时调用""" global external_api_session # 避免重复初始化 if external_api_session is not None: return session = requests.Session() # 配置Basic认证信息 session.auth = (settings.EXTERNAL_API_USERNAME, settings.EXTERNAL_API_PASSWORD) # 配置公共请求头 session.headers.update({ "Content-Type": "application/json", "Accept": "application/json" }) # 可选:配置重试策略,处理偶发网络波动 retry_config = requests.adapters.Retry( total=3, backoff_factor=1, allowed_methods=["GET"] ) http_adapter = requests.adapters.HTTPAdapter(max_retries=retry_config) session.mount("https://", http_adapter) session.mount("http://", http_adapter) external_api_session = session
3. 注册启动初始化钩子
在对应业务app的AppConfig中重写ready()方法,注册启动时的会话初始化逻辑:
# 你的业务app(比如负责元数据接口的app)的apps.py from django.apps import AppConfig class MetadataAppConfig(AppConfig): default_auto_field = "django.db.models.BigAutoField" name = "metadata" # 替换为你的app实际名称 def ready(self): # 应用启动完成后初始化第三方API会话 from utils.external_api_client import init_external_session init_external_session()
注意:不要在模块顶层直接初始化Session,Django启动过程中存在多次模块导入逻辑,可能导致重复初始化生成多个实例。
4. 业务代码中复用全局Session
替换原来视图里每次新建Session的逻辑,直接导入全局实例使用:
from django.conf import settings from utils.external_api_client import external_api_session from utils.external_api_client import init_external_session def metadata(jira_key: str, format=None): try: print(f"fetching {jira_key} with service account credentials") url = f"{settings.EXTERNAL_API_BASE_URL}/issue/{jira_key}" response = external_api_session.get(url) # 可选:遇到401认证失败时自动重新初始化会话,兼容凭证/会话过期场景 if response.status_code == 401: init_external_session() response = external_api_session.get(url) print(f"response: {response.status_code}") return response except Exception as e: message = {"error": "Uncaught error", "message": str(e)} return message
额外说明
- 复用Session后,底层会自动复用TCP长连接,同域名请求不需要重复做TCP握手、TLS协商,请求延迟会明显降低
- 不要手动调用
session.close(),全局实例会在进程生命周期内一直保持可用 - 如果第三方API有单会话请求频率限制,可以根据实际情况调整重试策略的间隔参数,避免触发限频
内容的提问来源于stack exchange,提问作者Austin Hallett
相关产品推荐
相关产品推荐

