Function wrapper

ew_decorator/with_timeout.py:25–50

(*args, **kwargs)

        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            seconds = default_seconds
            
            if timeout_param:
                # 先检查kwargs中是否有该参数
                if timeout_param in kwargs:
                    seconds = kwargs[timeout_param]
                else:
                    # 如果kwargs中没有，尝试从位置参数中获取
                    param_names = list(sig.parameters.keys())
                    if timeout_param in param_names:
                        param_index = param_names.index(timeout_param)
                        if param_index < len(args):
                            seconds = args[param_index]
                        else:
                            # 如果位置参数中也没有，尝试从函数默认参数中获取
                            param = sig.parameters[timeout_param]
                            if param.default != inspect.Parameter.empty:
                                seconds = param.default
            
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    return future.result(timeout=seconds)
                except concurrent.futures.TimeoutError:
                    raise TimeoutError(f"函数 {func.__name__} 执行超时（{seconds}秒）")
                
        return wrapper
    return decorator