如何在实例化前注入延迟逻辑,同时保留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
相关产品推荐
相关产品推荐

