如何缓存并复用带域范围委派的google.oauth2.service_account.Credentials对象？

我之前也碰到过这个头疼的问题——普通用户凭证的序列化/反序列化很方便，但服务账号带域委派的凭证确实没有现成的方法，总用pickle又担心安全和兼容性问题。给你分享几个亲测有效的替代方案：

方案一：手动提取核心字段，自定义序列化/反序列化 #

服务账号凭证的核心信息其实都可以从对象中直接获取，我们可以手动把这些字段提取出来转成字典，存到Redis里；需要复用的时候再从字典重建凭证对象，完全不用依赖pickle。

序列化（存到Redis） #

先写个函数把凭证对象转成可序列化的字典：

from datetime import datetime

def serialize_delegated_creds(creds):
    return {
        "token": creds.token,
        "token_uri": creds.token_uri,
        "client_id": creds.client_id,
        "client_secret": creds.client_secret,
        "scopes": creds.scopes,
        "subject": creds._subject,  # 这里是私有属性，但实际使用中稳定可访问
        "expiry": creds.expiry.isoformat() if creds.expiry else None
    }

# 使用示例：把序列化后的字典转成JSON存Redis
import json
redis.set("delegated_creds:user@domain.com", json.dumps(serialize_delegated_creds(creds)))

反序列化（从Redis加载） #

再写个函数把缓存的字典重新转成Credentials对象：

from google.oauth2.service_account import Credentials
from datetime import datetime

def deserialize_delegated_creds(serialized_dict):
    # 直接初始化服务账号凭证对象，传入缓存的核心参数
    creds = Credentials(
        token=serialized_dict["token"],
        token_uri=serialized_dict["token_uri"],
        client_id=serialized_dict["client_id"],
        client_secret=serialized_dict["client_secret"],
        scopes=serialized_dict["scopes"],
        subject=serialized_dict["subject"]
    )
    # 还原过期时间
    if serialized_dict["expiry"]:
        creds.expiry = datetime.fromisoformat(serialized_dict["expiry"])
    return creds

# 使用示例：从Redis取数据并反序列化
serialized_data = json.loads(redis.get("delegated_creds:user@domain.com"))
creds = deserialize_delegated_creds(serialized_data)

# 过期自动刷新：如果凭证过期，调用refresh方法会自动获取新token
if creds.expired:
    from google.auth.transport.requests import Request
    creds.refresh(Request())
    # 记得把刷新后的凭证重新存回Redis哦

注意点：

• _subject是私有属性，但目前google-auth版本中这个字段稳定可用；如果担心未来版本变化，建议在首次创建委派凭证时就把subject单独存下来，不用从对象中提取。
• 一定要处理过期时间，不然凭证会被判定为永远有效，无法自动刷新。

方案二：用安全的序列化工具替代pickle #

如果实在不想手动写序列化逻辑，可以用更安全的序列化工具代替标准pickle，比如cloudpickle——它能处理更多Python对象，而且比标准pickle安全一些（当然前提是你的Redis是可信环境，没有恶意数据）。

示例代码：

import cloudpickle
import redis

# 存到Redis
redis.set("delegated_creds:user@domain.com", cloudpickle.dumps(creds))

# 从Redis加载
creds = cloudpickle.loads(redis.get("delegated_creds:user@domain.com"))

注意点：

• 虽然cloudpickle比pickle安全，但仍然要确保Redis的访问权限严格控制，不要存入不可信数据。
• 不同Python版本或google-auth版本之间可能存在兼容性问题，升级依赖时要测试。

方案三：利用google-auth的内置缓存（仅限单进程场景） #

如果你的应用是单进程部署，也可以直接用google-auth内置的缓存机制，不用自己操作Redis。比如创建凭证时指定缓存对象：

from google.auth.transport.requests import Request
from google.oauth2.service_account import Credentials
import google.auth.cache

# 用内存缓存（单进程有效）
cache = google.auth.cache.MemoryCache()

creds = Credentials.from_service_account_file(
    "service-account-key.json",
    scopes=["https://www.googleapis.com/auth/admin.directory.user"],
    subject="user@domain.com"
)
# 绑定缓存，后续刷新会自动复用缓存的凭证
creds.refresh(Request(cache=cache))

不过这个方案只适合单进程，跨进程或分布式部署的话还是得用前两种方法。

总结来说，方案一是最推荐的——完全可控、无安全风险，而且能灵活适配Redis这类缓存工具，不用担心版本兼容性问题。

备注：内容来源于stack exchange，提问作者Héctor H. Hache