You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在实例化前注入延迟逻辑,同时保留IsAllowed(perms=[IsAdmin(runtime_value)|HasPerm('*')])的调用接口?

当然可以实现你想要的简洁接口!我们可以通过几种方式把perms参数的计算延迟到__call__阶段,同时保持IsAllowed(perms=[IsAdmin(runtime_value) | HasPerm('*')])的调用形式。下面是几个可行的实现思路:

思路1:利用元类捕获初始化表达式

元类可以拦截IsAllowed的实例化过程,我们可以借助它捕获传入perms的原始表达式,再把表达式包装成延迟执行的函数,等到调用__call__时再传入runtime_value计算结果。

import ast
import inspect

class DelayedEvalMeta(type):
    def __call__(cls, *args, **kwargs):
        # 获取调用IsAllowed的上下文源代码
        caller_frame = inspect.currentframe().f_back
        try:
            source = inspect.getsource(caller_frame)
            # 解析AST找到perms参数的表达式
            tree = ast.parse(source)
            for node in ast.walk(tree):
                if isinstance(node, ast.Call) and getattr(node.func, 'id', '') == 'IsAllowed':
                    for kw in node.keywords:
                        if kw.arg == 'perms':
                            # 把表达式转为字符串,包装成带runtime_value参数的lambda
                            expr_str = ast.unparse(kw.value)
                            delayed_func = eval(f"lambda runtime_value: {expr_str}", globals())
                            kwargs['perms'] = delayed_func
                            break
                    break
        finally:
            # 确保释放帧引用,避免内存泄漏
            del caller_frame
        return super().__call__(*args, **kwargs)

class IsAllowed(object, metaclass=DelayedEvalMeta):
    def __init__(self, perms=None):
        self.perms = perms
    def __call__(self, runtime_value):
        return self.perms(runtime_value)

# 完全符合你想要的调用形式
checker = IsAllowed(perms=[IsAdmin(runtime_value) | HasPerm('*')])
# 后续传入runtime_value执行计算
result = checker(some_runtime_user)

注意:这个方案依赖inspect和AST解析,需要确保调用IsAllowed的代码是可被inspect读取的(比如不能是交互式环境或混淆压缩后的代码),AST解析逻辑也需要根据实际调用场景调整(比如处理位置参数的情况)。

思路2:用装饰器增强初始化逻辑

如果觉得元类太重,你可以用装饰器来替换IsAllowed的__init__方法,在初始化时捕获perms的表达式并转为延迟函数,逻辑和元类类似但更轻量。

import ast
import inspect

def delayed_perms(cls):
    original_init = cls.__init__
    def new_init(self, *args, **kwargs):
        if 'perms' in kwargs:
            caller_frame = inspect.currentframe().f_back
            try:
                source = inspect.getsource(caller_frame)
                tree = ast.parse(source)
                for node in ast.walk(tree):
                    if isinstance(node, ast.Call) and getattr(node.func, 'id', '') == 'IsAllowed':
                        for kw in node.keywords:
                            if kw.arg == 'perms':
                                expr_str = ast.unparse(kw.value)
                                kwargs['perms'] = eval(f"lambda runtime_value: {expr_str}", globals())
                                break
                        break
            finally:
                del caller_frame
        original_init(self, *args, **kwargs)
    cls.__init__ = new_init
    return cls

@delayed_perms
class IsAllowed(object):
    def __init__(self, perms=None):
        self.perms = perms
    def __call__(self, runtime_value):
        return self.perms(runtime_value)

# 同样支持你期望的调用方式
checker = IsAllowed(perms=[IsAdmin(runtime_value) | HasPerm('*')])
result = checker(some_runtime_user)

思路3:自定义可延迟计算的权限对象(更可控)

如果你想避免依赖inspect和AST,可以让IsAdmin、HasPerm返回可延迟计算的对象,通过重载运算符把整个表达式转换成可执行的逻辑,等到__call__时再传入runtime_value计算。

class DelayedCondition:
    def __init__(self, evaluate):
        self.evaluate = evaluate
    
    def __or__(self, other):
        # 组合两个条件,实现逻辑或
        def combined_eval(runtime_value):
            return self.evaluate(runtime_value) or other.evaluate(runtime_value)
        return DelayedCondition(combined_eval)

class IsAdmin(DelayedCondition):
    def __init__(self):
        # 不提前传入runtime_value,而是在evaluate中接收
        super().__init__(lambda runtime_value: runtime_value.is_admin)

class HasPerm(DelayedCondition):
    def __init__(self, perm):
        super().__init__(lambda runtime_value: perm in runtime_value.permissions)

class IsAllowed(object):
    def __init__(self, perms=None):
        self.perms = perms
    def __call__(self, runtime_value):
        # 遍历所有延迟条件,执行计算
        return any(cond.evaluate(runtime_value) for cond in self.perms)

# 调用方式接近你的需求,只是IsAdmin不需要传参数
checker = IsAllowed(perms=[IsAdmin() | HasPerm('*')])
result = checker(some_runtime_user)

这个方案更可控,不依赖外部模块,只是需要调整IsAdmin的调用方式(不需要提前传runtime_value)。

内容的提问来源于stack exchange,提问作者Amir Afianian

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.27 20:17:30