import functools
from typing import Any, Callable, TypeVar
T = TypeVar('T') # 定义泛型
def timeout(after: float, action: Callable[[], None]) -> Callable[[T], T]:
"""
after：超时时间（秒）
action：超时后执行的回调函数
"""
def decorator(func: T) -> T:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
class TimeoutError(Exception):
pass
def _timeout_handler() -> None:
raise TimeoutError()
import threading
thread = threading.Thread(target=_timeout_handler)
thread.start()
try:
thread.join(after)
return func(*args, **kwargs)
except TimeoutError:
action()
raise
return wrapper
return decorator
# 使用示例
@timeout(2, lambda: print("函数执行超时，进行了回调操作！"))
def long_running_function() -> None:
import time
time.sleep(3)
long_running_function()这段代码定义了一个timeout装饰器，用于处理被装饰的函数执行时超时的情况。它接受两个参数：after是超时的秒数，action是超时后要执行的回调函数。如果函数执行时间超过after秒，则会触发超时异常，并执行指定的回调函数。这个装饰器使用了线程来处理超时逻辑，确保主线程不会被阻塞。