Python全局变量最佳实践:跨文件仅执行一次f_token复用v_Token
多独立Python脚本复用Token的最佳实践
针对你需要在多个独立执行的脚本中只调用一次f_token()获取Token并复用的需求,以下是几种实用的最佳实践方案:
方案1:本地文件缓存Token(最简单的跨脚本复用方式)
通过将Token和过期信息存储到本地文件,每次运行脚本时先检查缓存是否有效,有效则直接使用,无效再重新获取。
修改00.py代码(加入缓存逻辑)
import json import time import adal # 建议将敏感配置放到环境变量或加密配置文件,不要硬编码 AUTH_URL = "https://login.microsoftonline.com/common" RESOURCE_URL = "https://xxx.xxx.dynamics.com/" USERNAME = "your_username" PASSWORD = "your_password" CLIENT_ID = "your_client_id" CACHE_FILE = "token_cache.json" def get_cached_token(): """读取本地缓存的有效Token""" try: with open(CACHE_FILE, 'r') as f: cache_data = json.load(f) # 根据实际返回的token_response字段调整过期判断逻辑 expires_on = float(cache_data.get('expiresOn', 0)) # 提前60秒刷新,避免Token在使用时过期 if time.time() < expires_on - 60: return cache_data['accessToken'] except (FileNotFoundError, json.JSONDecodeError): return None def save_token_to_cache(token_response): """将Token和过期信息写入缓存文件""" with open(CACHE_FILE, 'w') as f: json.dump(token_response, f) def f_token(): # 优先尝试获取缓存的有效Token cached_token = get_cached_token() if cached_token: return cached_token # 缓存无效时重新获取Token auth_context = adal.AuthenticationContext(AUTH_URL) token_response = auth_context.acquire_token_with_username_password( RESOURCE_URL, USERNAME, PASSWORD, CLIENT_ID ) save_token_to_cache(token_response) return token_response['accessToken']
其他脚本(01/02/03.py)调用方式
保持原有导入逻辑即可,f_token()会自动处理缓存判断:
from 00 import f_token A_Token = f_token() def f_meta(): # 使用A_Token的业务代码
注意事项:
- 敏感信息(用户名、密码等)不要硬编码,建议用
os.getenv()读取环境变量。 - 缓存文件需设置合适的权限,避免敏感信息泄露。
方案2:Redis缓存(适合多进程/多机器共享场景)
如果你的脚本需要在多台机器或多个进程中频繁运行,本地文件缓存灵活性不足,可以用Redis实现跨进程、跨机器的Token共享。
第一步:安装依赖
pip install redis
修改00.py代码(Redis缓存逻辑)
import time import adal import redis AUTH_URL = "https://login.microsoftonline.com/common" RESOURCE_URL = "https://xxx.xxx.dynamics.com/" USERNAME = "your_username" PASSWORD = "your_password" CLIENT_ID = "your_client_id" REDIS_HOST = "localhost" REDIS_PORT = 6379 REDIS_KEY = "dynamics_token" # 初始化Redis客户端 redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0) def get_cached_token(): """从Redis读取有效Token""" token_data = redis_client.hgetall(REDIS_KEY) if not token_data: return None access_token = token_data.get(b'accessToken').decode('utf-8') expires_on = float(token_data.get(b'expiresOn').decode('utf-8')) if time.time() < expires_on - 60: return access_token return None def save_token_to_cache(token_response): """将Token存入Redis,并设置过期时间""" expires_on = float(token_response['expiresOn']) redis_client.hset(REDIS_KEY, mapping={ 'accessToken': token_response['accessToken'], 'expiresOn': str(expires_on) }) # 设置Redis键的过期时间,比Token实际过期时间早60秒 redis_client.expire(REDIS_KEY, int(expires_on - time.time() - 60)) def f_token(): cached_token = get_cached_token() if cached_token: return cached_token auth_context = adal.AuthenticationContext(AUTH_URL) token_response = auth_context.acquire_token_with_username_password( RESOURCE_URL, USERNAME, PASSWORD, CLIENT_ID ) save_token_to_cache(token_response) return token_response['accessToken']
其他脚本调用方式和方案1一致,所有脚本都会共享Redis中的Token缓存。
方案3:主脚本统一获取Token(适合批量执行场景)
如果三个脚本是通过主脚本批量触发执行的,而非单独手动运行,可以让主脚本先获取Token,再通过环境变量传递给子脚本。
主脚本(main.py)
import subprocess from 00 import f_token # 统一获取一次Token token = f_token() # 通过环境变量将Token传递给各个子脚本 subprocess.run(["python", "01.py"], env={"DYNAMICS_TOKEN": token}) subprocess.run(["python", "02.py"], env={"DYNAMICS_TOKEN": token}) subprocess.run(["python", "03.py"], env={"DYNAMICS_TOKEN": token})
修改子脚本(01/02/03.py)
import os # 从环境变量读取Token A_Token = os.getenv("DYNAMICS_TOKEN") def f_meta(): # 使用A_Token的业务代码
这种方式确保Token只获取一次,适合需要批量调度脚本的场景。
内容的提问来源于stack exchange,提问作者samronaldo309
相关产品推荐
相关产品推荐

