如何使用Python装饰器实现会话验证？附会话类示例代码

嘿，我来帮你搞定这个会话验证的需求！先看你给出的Session类代码（看起来你截断了@staticmethod部分），我先把它补全，然后一步步实现装饰器。

先补全你的Session类 #

首先，我假设get_token_details是个静态方法，还新增了一个refresh_token方法——比起每次重新实例化Session，直接刷新token更高效：

import datetime

class Session:
    # 先补上你提到的类属性，用来存获取token的payload和headers
    _token_payload = {}
    _token_headers = {}

    def __init__(self):
        self._token_response_dict = self.get_token_details(Session._token_payload, Session._token_headers)
        self.token = self._token_response_dict['access_token']
        self.token_valid_time = datetime.datetime.now() + datetime.timedelta(
            seconds=self._token_response_dict['expires_in']
        )
        self.token_type = self._token_response_dict['token_type']

    def is_valid(self):
        return datetime.datetime.now() < self.token_valid_time

    @staticmethod
    def get_token_details(payload, headers):
        # 这里替换成你真实的获取token的接口请求逻辑
        # 比如用requests.post(url, json=payload, headers=headers).json()
        return {
            'access_token': 'dummy_token_123',
            'expires_in': 3600,
            'token_type': 'Bearer'
        }
    
    # 新增刷新token的方法，避免重复实例化
    def refresh_token(self):
        self._token_response_dict = self.get_token_details(Session._token_payload, Session._token_headers)
        self.token = self._token_response_dict['access_token']
        self.token_valid_time = datetime.datetime.now() + datetime.timedelta(
            seconds=self._token_response_dict['expires_in']
        )
        self.token_type = self._token_response_dict['token_type']

实现会话验证装饰器 #

装饰器的核心逻辑很简单：在执行目标函数前，先检查会话是否有效；如果无效，就刷新token，再执行原函数。我给你两种常用的实现方式：

方式1：支持多会话的装饰器（更灵活） #

这种方式允许你给不同的函数传递不同的Session实例，适合多用户/多场景的情况：

def session_required(session):
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 检查会话有效性
            if not session.is_valid():
                print("会话已过期，正在自动刷新token...")
                session.refresh_token()
            # 把会话实例传给目标函数，方便后续使用
            return func(*args, **kwargs, session=session)
        return wrapper
    return decorator

使用示例

# 创建一个会话实例
my_session = Session()

# 给需要验证的函数加上装饰器
@session_required(my_session)
def call_protected_api(session, *args):
    print(f"调用受保护API，使用token: {session.token}")
    # 这里写你的真实API请求逻辑，比如用session.token设置请求头

# 第一次调用，会话有效，直接执行
call_protected_api()

# 模拟token过期
my_session.token_valid_time = datetime.datetime.now() - datetime.timedelta(hours=1)
# 再次调用，会自动刷新token后再执行
call_protected_api()

方式2：全局单例会话的装饰器（更简洁） #

如果你的应用里只有一个全局会话（比如单用户后台），可以用这种简化版：

# 先创建全局会话实例
global_session = Session()

def global_session_required(func):
    def wrapper(*args, **kwargs):
        if not global_session.is_valid():
            print("全局会话过期，正在刷新token...")
            global_session.refresh_token()
        return func(*args, **kwargs, session=global_session)
    return wrapper

# 使用示例
@global_session_required
def another_protected_call(session):
    print(f"全局会话调用API，token: {session.token}")

another_protected_call()

一些实用提示 #

• 线程安全：如果是多线程环境，记得给refresh_token方法加锁（比如用threading.Lock()），避免多个线程同时触发刷新，导致重复请求token。
• 异常处理：在get_token_details和refresh_token里加上异常捕获（比如请求超时、接口报错），可以加个重试逻辑，避免装饰器抛出异常导致业务函数挂掉。
• 自定义传递逻辑：你可以根据需求调整装饰器，比如直接把token作为参数传给函数，而不是整个会话实例。

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