mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
refactor(types): describe sync and async debounce contracts
This commit is contained in:
@@ -3,19 +3,45 @@ import functools
|
||||
import inspect
|
||||
from abc import ABC, abstractmethod
|
||||
from threading import Lock, Timer
|
||||
from typing import Any, Callable, Optional
|
||||
from typing import Awaitable, Callable, Concatenate, Generic, Optional, ParamSpec, Protocol, TypeVar, cast, overload
|
||||
|
||||
from app.runtime.log import logger
|
||||
|
||||
_P = ParamSpec("_P")
|
||||
_R = TypeVar("_R")
|
||||
_T = TypeVar("_T")
|
||||
|
||||
class BaseDebouncer(ABC):
|
||||
|
||||
class _Debounced(Protocol[_P]):
|
||||
"""保留调用参数;实际函数种类决定调用与取消是否返回可等待对象。"""
|
||||
|
||||
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> Awaitable[None] | None:
|
||||
"""提交本次调用;同步函数直接返回,协程函数需要等待提交。"""
|
||||
...
|
||||
|
||||
# 包装函数挂载的控制方法,异步引擎会等待任务进入终态。
|
||||
cancel: Callable[[], Awaitable[None] | None]
|
||||
|
||||
@overload
|
||||
def __get__(self, instance: None, owner: type[object] | None = None) -> "_Debounced[_P]":
|
||||
"""从类访问时保留尚未绑定的参数。"""
|
||||
...
|
||||
|
||||
@overload
|
||||
def __get__(self: "_Debounced[Concatenate[_T, _P]]", instance: _T,
|
||||
owner: type[_T] | None = None) -> "_Debounced[_P]":
|
||||
"""从实例访问时由函数描述符绑定首个参数。"""
|
||||
...
|
||||
|
||||
|
||||
class BaseDebouncer(ABC, Generic[_P, _R]):
|
||||
"""
|
||||
防抖器的抽象基类。定义了防抖器的基本接口和日志功能。
|
||||
所有防抖器实现类必须继承此类并实现其抽象方法。
|
||||
"""
|
||||
|
||||
def __init__(self, func: Callable, interval: float, *,
|
||||
leading: bool = False, enable_logging: bool = False, source: str = ""):
|
||||
def __init__(self, func: Callable[_P, _R], interval: float, *,
|
||||
leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
|
||||
"""
|
||||
初始化防抖器实例。
|
||||
:param func: 要防抖的函数或协程
|
||||
@@ -31,14 +57,14 @@ class BaseDebouncer(ABC):
|
||||
self.source = source
|
||||
|
||||
@abstractmethod
|
||||
def __call__(self, *args, **kwargs) -> None:
|
||||
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> Awaitable[None] | None:
|
||||
"""
|
||||
定义防抖调用的契约,子类必须实现。
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def cancel(self) -> None:
|
||||
def cancel(self) -> Awaitable[None] | None:
|
||||
"""
|
||||
定义取消挂起调用的契约,子类必须实现。
|
||||
"""
|
||||
@@ -50,7 +76,7 @@ class BaseDebouncer(ABC):
|
||||
"""
|
||||
return f"[{self.source}] {message}" if self.source else message
|
||||
|
||||
def log(self, level: str, message: str):
|
||||
def log(self, level: str, message: str) -> None:
|
||||
"""
|
||||
根据日志级别记录日志。
|
||||
"""
|
||||
@@ -58,53 +84,54 @@ class BaseDebouncer(ABC):
|
||||
log_method = getattr(logger, level, logger.debug)
|
||||
log_method(self.format_log(message))
|
||||
|
||||
def log_debug(self, message: str):
|
||||
def log_debug(self, message: str) -> None:
|
||||
"""
|
||||
记录调试日志。
|
||||
"""
|
||||
self.log("debug", message)
|
||||
|
||||
def log_info(self, message: str):
|
||||
def log_info(self, message: str) -> None:
|
||||
"""
|
||||
记录信息日志。
|
||||
"""
|
||||
self.log("info", message)
|
||||
|
||||
def log_warning(self, message: str):
|
||||
def log_warning(self, message: str) -> None:
|
||||
"""
|
||||
记录警告日志。
|
||||
"""
|
||||
self.log("warning", message)
|
||||
|
||||
def error(self, message: str):
|
||||
def error(self, message: str) -> None:
|
||||
"""
|
||||
记录错误日志。
|
||||
"""
|
||||
self.log("error", message)
|
||||
|
||||
def critical(self, message: str):
|
||||
def critical(self, message: str) -> None:
|
||||
"""
|
||||
记录严重错误日志。
|
||||
"""
|
||||
self.log("critical", message)
|
||||
|
||||
|
||||
class Debouncer(BaseDebouncer):
|
||||
class Debouncer(BaseDebouncer[_P, object]):
|
||||
"""
|
||||
同步防抖实现类
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
def __init__(self, func: Callable[_P, object], interval: float, *,
|
||||
leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
|
||||
"""
|
||||
初始化防抖器实例。
|
||||
"""
|
||||
super().__init__(*args, **kwargs)
|
||||
super().__init__(func, interval, leading=leading, enable_logging=enable_logging, source=source)
|
||||
self.timer: Optional[Timer] = None
|
||||
self.lock = Lock()
|
||||
# 用于前沿模式,标记是否处于“冷却”或“不应期”
|
||||
self.is_cooling_down = False
|
||||
|
||||
def __call__(self, *args, **kwargs) -> None:
|
||||
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
调用防抖函数。
|
||||
:param args:
|
||||
@@ -117,7 +144,7 @@ class Debouncer(BaseDebouncer):
|
||||
else:
|
||||
self._call_trailing(*args, **kwargs)
|
||||
|
||||
def _call_leading(self, *args, **kwargs):
|
||||
def _call_leading(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
前沿模式的逻辑。
|
||||
"""
|
||||
@@ -138,7 +165,7 @@ class Debouncer(BaseDebouncer):
|
||||
self.timer.start()
|
||||
self.log_debug(f"前沿模式: 进入 {self.interval} 秒的冷却期。")
|
||||
|
||||
def _end_cool_down(self):
|
||||
def _end_cool_down(self) -> None:
|
||||
"""
|
||||
计时器到期后,解除冷却状态
|
||||
"""
|
||||
@@ -146,7 +173,7 @@ class Debouncer(BaseDebouncer):
|
||||
self.is_cooling_down = False
|
||||
self.log_debug("前沿模式: 冷却时间结束,可以再次立即执行。")
|
||||
|
||||
def _call_trailing(self, *args, **kwargs):
|
||||
def _call_trailing(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
后沿模式的逻辑。
|
||||
"""
|
||||
@@ -155,7 +182,8 @@ class Debouncer(BaseDebouncer):
|
||||
self.timer.cancel()
|
||||
self.log_debug("后沿模式: 检测到新的调用,已重置计时器。")
|
||||
|
||||
def execute():
|
||||
def execute() -> None:
|
||||
"""计时结束后执行最后一次调用保存的参数。"""
|
||||
self.log_info("后沿模式: 计时结束,开始执行函数。")
|
||||
self.func(*args, **kwargs)
|
||||
|
||||
@@ -175,18 +203,19 @@ class Debouncer(BaseDebouncer):
|
||||
self.is_cooling_down = False
|
||||
|
||||
|
||||
class AsyncDebouncer(BaseDebouncer):
|
||||
class AsyncDebouncer(BaseDebouncer[_P, Awaitable[object]]):
|
||||
"""
|
||||
异步防抖实现类。
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
def __init__(self, func: Callable[_P, Awaitable[object]], interval: float, *,
|
||||
leading: bool = False, enable_logging: bool = False, source: str = "") -> None:
|
||||
"""
|
||||
初始化异步防抖器实例。
|
||||
"""
|
||||
super().__init__(*args, **kwargs)
|
||||
self.task: Optional[asyncio.Task] = None
|
||||
self._retired_tasks: set[asyncio.Task[Any]] = set()
|
||||
super().__init__(func, interval, leading=leading, enable_logging=enable_logging, source=source)
|
||||
self.task: Optional[asyncio.Task[None]] = None
|
||||
self._retired_tasks: set[asyncio.Task[None]] = set()
|
||||
self.lock = asyncio.Lock()
|
||||
self.is_cooling_down = False
|
||||
|
||||
@@ -199,7 +228,7 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
self._retired_tasks.add(task)
|
||||
task.add_done_callback(self._retired_tasks.discard)
|
||||
|
||||
async def __call__(self, *args, **kwargs) -> None:
|
||||
async def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
异步调用防抖函数。
|
||||
"""
|
||||
@@ -209,7 +238,7 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
else:
|
||||
await self._call_trailing(*args, **kwargs)
|
||||
|
||||
async def _call_leading(self, *args, **kwargs):
|
||||
async def _call_leading(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
前沿模式的逻辑。
|
||||
"""
|
||||
@@ -223,7 +252,7 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
self.task = asyncio.create_task(self._end_cool_down())
|
||||
self.log_debug(f"前沿模式 (async): 进入 {self.interval} 秒的冷却期。")
|
||||
|
||||
async def _end_cool_down(self):
|
||||
async def _end_cool_down(self) -> None:
|
||||
"""
|
||||
计时器到期后,解除冷却状态
|
||||
"""
|
||||
@@ -232,7 +261,7 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
self.is_cooling_down = False
|
||||
self.log_debug("前沿模式 (async): 冷却时间结束。")
|
||||
|
||||
async def _call_trailing(self, *args, **kwargs):
|
||||
async def _call_trailing(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
后沿模式的逻辑。
|
||||
"""
|
||||
@@ -243,7 +272,7 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
self.task = asyncio.create_task(self._delayed_execute(*args, **kwargs))
|
||||
self.log_debug(f"后沿模式 (async): 任务已创建,将在 {self.interval} 秒后执行。")
|
||||
|
||||
async def _delayed_execute(self, *args, **kwargs):
|
||||
async def _delayed_execute(self, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""
|
||||
延迟执行实际的协程函数。
|
||||
"""
|
||||
@@ -279,26 +308,29 @@ class AsyncDebouncer(BaseDebouncer):
|
||||
|
||||
|
||||
def debounce(interval: float, *, leading: bool = False,
|
||||
enable_logging: bool = False, source: str = "") -> Callable:
|
||||
enable_logging: bool = False, source: str = "") -> Callable[[Callable[_P, object]], _Debounced[_P]]:
|
||||
"""
|
||||
支持同步和异步的防抖装饰器工厂。
|
||||
"""
|
||||
|
||||
def decorator(func: Callable) -> Callable:
|
||||
def decorator(func: Callable[_P, object]) -> _Debounced[_P]:
|
||||
"""按实际协程函数标记选择引擎,保留原函数的参数与绑定行为。"""
|
||||
# 检查函数类型,并选择合适的引擎
|
||||
if inspect.iscoroutinefunction(func):
|
||||
# 异步函数,使用 AsyncDebouncer
|
||||
instance = AsyncDebouncer(func, interval,
|
||||
leading=leading,
|
||||
enable_logging=enable_logging,
|
||||
source=source)
|
||||
async_instance = AsyncDebouncer(func, interval,
|
||||
leading=leading,
|
||||
enable_logging=enable_logging,
|
||||
source=source)
|
||||
|
||||
@functools.wraps(func)
|
||||
async def async_wrapper(*args, **kwargs) -> Any:
|
||||
await instance(*args, **kwargs)
|
||||
async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""等待异步引擎接收本次调用。"""
|
||||
await async_instance(*args, **kwargs)
|
||||
|
||||
async_wrapper.cancel = instance.cancel
|
||||
return async_wrapper
|
||||
async_result = cast(_Debounced[_P], async_wrapper)
|
||||
async_result.cancel = async_instance.cancel
|
||||
return async_result
|
||||
|
||||
else:
|
||||
# 同步函数,使用 Debouncer
|
||||
@@ -308,10 +340,12 @@ def debounce(interval: float, *, leading: bool = False,
|
||||
source=source)
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs) -> Any:
|
||||
def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
|
||||
"""通过同步引擎接收本次调用。"""
|
||||
instance(*args, **kwargs)
|
||||
|
||||
wrapper.cancel = instance.cancel
|
||||
return wrapper
|
||||
result = cast(_Debounced[_P], wrapper)
|
||||
result.cancel = instance.cancel
|
||||
return result
|
||||
|
||||
return decorator
|
||||
|
||||
184
tests/test_debounce_contract.py
Normal file
184
tests/test_debounce_contract.py
Normal file
@@ -0,0 +1,184 @@
|
||||
"""防抖器参数转发、函数绑定与同步异步选择的兼容契约。"""
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
|
||||
import pytest
|
||||
|
||||
import app.runtime.debounce as debounce_module
|
||||
from app.runtime.debounce import AsyncDebouncer, Debouncer, debounce
|
||||
|
||||
|
||||
class _ManualTimer:
|
||||
"""由测试显式推进的定时器,避免线程与墙钟时间影响断言。"""
|
||||
|
||||
def __init__(self, interval, callback):
|
||||
"""保存真实引擎提供的间隔与回调。"""
|
||||
self.interval = interval
|
||||
self.callback = callback
|
||||
self.active = False
|
||||
|
||||
def start(self):
|
||||
"""标记为待执行。"""
|
||||
self.active = True
|
||||
|
||||
def cancel(self):
|
||||
"""撤销尚未执行的回调。"""
|
||||
self.active = False
|
||||
|
||||
def is_alive(self):
|
||||
"""提供与线程定时器一致的存活查询。"""
|
||||
return self.active
|
||||
|
||||
def fire(self):
|
||||
"""仅执行仍有效的回调。"""
|
||||
if self.active:
|
||||
self.active = False
|
||||
self.callback()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def manual_timers(monkeypatch):
|
||||
"""捕获引擎创建的定时器并禁止启动真实线程。"""
|
||||
timers = []
|
||||
|
||||
def create_timer(interval, callback):
|
||||
"""返回可由测试控制的独立定时器。"""
|
||||
timer = _ManualTimer(interval, callback)
|
||||
timers.append(timer)
|
||||
return timer
|
||||
|
||||
monkeypatch.setattr(debounce_module, "Timer", create_timer)
|
||||
return timers
|
||||
|
||||
|
||||
@pytest.mark.parametrize("keyword_constructor", [False, True])
|
||||
@pytest.mark.parametrize("leading", [False, True])
|
||||
def test_sync_debouncer_preserves_arguments_and_cooling(
|
||||
manual_timers, keyword_constructor, leading
|
||||
):
|
||||
"""两种构造方式均保留参数,并分别执行首个调用或最后一个调用。"""
|
||||
calls = []
|
||||
|
||||
def record(value, *, label):
|
||||
"""保留位置参数、关键字参数和对象身份。"""
|
||||
calls.append((value, label))
|
||||
return "discarded"
|
||||
|
||||
options = {"leading": leading, "enable_logging": False, "source": "test"}
|
||||
engine = (
|
||||
Debouncer(func=record, interval=7, **options)
|
||||
if keyword_constructor
|
||||
else Debouncer(record, 7, **options)
|
||||
)
|
||||
first, last = object(), object()
|
||||
assert engine(first, label="first") is None
|
||||
assert engine(last, label="last") is None
|
||||
assert manual_timers[0].is_alive() is False
|
||||
assert manual_timers[-1].interval == 7
|
||||
assert calls == ([(first, "first")] if leading else [])
|
||||
for timer in manual_timers:
|
||||
timer.fire()
|
||||
assert calls == ([(first, "first")] if leading else [(last, "last")])
|
||||
assert engine.is_cooling_down is False
|
||||
engine(first, label="cancelled")
|
||||
before_cancel = list(calls)
|
||||
assert engine.cancel() is None
|
||||
manual_timers[-1].fire()
|
||||
assert calls == before_cancel
|
||||
assert engine.timer is None
|
||||
|
||||
|
||||
def test_decorator_preserves_method_binding_and_metadata(manual_timers):
|
||||
"""普通函数描述符仍绑定实例,cancel 仍操作原先共享的防抖引擎。"""
|
||||
calls = []
|
||||
|
||||
class Receiver:
|
||||
"""通过实例方法暴露防抖包装器。"""
|
||||
|
||||
@debounce(3)
|
||||
def accept(self, value, *, label):
|
||||
"""记录接收实例与完整调用参数。"""
|
||||
calls.append((self, value, label))
|
||||
|
||||
receiver = Receiver()
|
||||
assert receiver.accept("value", label="label") is None
|
||||
assert Receiver.accept.__name__ == "accept"
|
||||
assert inspect.iscoroutinefunction(receiver.accept) is False
|
||||
manual_timers[-1].fire()
|
||||
assert calls == [(receiver, "value", "label")]
|
||||
receiver.accept("cancelled", label="label")
|
||||
assert receiver.accept.cancel() is None
|
||||
manual_timers[-1].fire()
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("leading", [False, True])
|
||||
async def test_async_debouncer_preserves_arguments_and_result_discard(leading):
|
||||
"""协程函数通过异步引擎执行,调用和取消均保持返回 None 的约定。"""
|
||||
calls = []
|
||||
|
||||
async def record(value, *, label):
|
||||
"""记录实际被执行的调用。"""
|
||||
calls.append((value, label))
|
||||
return "discarded"
|
||||
|
||||
engine = AsyncDebouncer(func=record, interval=0, leading=leading, source="test")
|
||||
assert await engine("first", label=1) is None
|
||||
assert await engine("last", label=2) is None
|
||||
await asyncio.wait_for(engine.task, timeout=1)
|
||||
assert calls == ([("first", 1)] if leading else [("last", 2)])
|
||||
assert await engine.cancel() is None
|
||||
assert engine.task is None
|
||||
assert engine._retired_tasks == set()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_decorator_preserves_binding_and_cancel():
|
||||
"""异步方法包装保留 await、实例绑定、元数据和异步取消接口。"""
|
||||
calls = []
|
||||
|
||||
class Receiver:
|
||||
"""提供异步防抖方法。"""
|
||||
|
||||
@debounce(60, leading=True)
|
||||
async def accept(self, value, *, label):
|
||||
"""记录绑定后的调用并返回被包装器丢弃的结果。"""
|
||||
calls.append((self, value, label))
|
||||
return "discarded"
|
||||
|
||||
receiver = Receiver()
|
||||
try:
|
||||
assert inspect.iscoroutinefunction(receiver.accept)
|
||||
assert receiver.accept.__name__ == "accept"
|
||||
assert await receiver.accept("value", label="label") is None
|
||||
assert calls == [(receiver, "value", "label")]
|
||||
finally:
|
||||
assert await receiver.accept.cancel() is None
|
||||
|
||||
|
||||
def test_sync_function_returning_awaitable_stays_sync(manual_timers):
|
||||
"""返回可等待对象的普通函数仍走同步引擎,不能仅凭返回类型改选异步引擎。"""
|
||||
events = []
|
||||
|
||||
class Result:
|
||||
"""记录是否有人等待了普通函数的返回值。"""
|
||||
|
||||
def __await__(self):
|
||||
"""提供不会实际等待的 await 协议。"""
|
||||
events.append("awaited")
|
||||
yield from ()
|
||||
return "result"
|
||||
|
||||
@debounce(3, leading=True)
|
||||
def callback(value):
|
||||
"""返回可等待对象但保持普通函数定义。"""
|
||||
events.append(value)
|
||||
return Result()
|
||||
|
||||
assert inspect.iscoroutinefunction(callback) is False
|
||||
assert callback("called") is None
|
||||
assert events == ["called"]
|
||||
assert callback.cancel() is None
|
||||
assert manual_timers[-1].is_alive() is False
|
||||
Reference in New Issue
Block a user