mirror of
https://github.com/A-Minos/nonebot-plugin-tetris-stats.git
synced 2026-03-05 05:36:54 +08:00
c8832bd1c9
* 🔧 启用一些 ruff 的新规则 * ➕ 添加开发依赖 nonebot-adapter-qq * ➕ 添加依赖 nonebot-plugin-session * ➕ 添加依赖 nonebot-plugin-session-orm * 🔧 忽略 ruff 规则 ISC001 format 冲突风险 * 🚨 修复 ruff 警报 * ♻️ 重构！ * ♻️ 恢复定时获取 TetraLeague 数据的功能 * ✨ 统一处理需要捕获的错误 * ✨ 记录用户触发的指令
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
from asyncio import sleep
|
|
from collections.abc import Awaitable, Callable
|
|
from datetime import timedelta
|
|
from functools import wraps
|
|
from typing import TypeVar, cast
|
|
|
|
from nonebot.log import logger
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
def retry(
|
|
max_attempts: int = 3,
|
|
exception_type: type[BaseException] | tuple[type[BaseException], ...] = Exception,
|
|
delay: timedelta | None = None,
|
|
) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
|
|
def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
|
|
@wraps(func)
|
|
async def wrapper(*args, **kwargs) -> T: # noqa: ANN002, ANN003
|
|
attempts = 0
|
|
while attempts < max_attempts + 1:
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except exception_type as e: # noqa: PERF203
|
|
logger.exception(e)
|
|
attempts += 1
|
|
if attempts <= max_attempts:
|
|
if delay is not None:
|
|
await sleep(delay.total_seconds())
|
|
logger.debug(f'Retrying: {func.__name__} ({attempts}/{max_attempts})')
|
|
continue
|
|
raise
|
|
msg = 'Unexpectedly reached the end of the retry loop'
|
|
raise RuntimeError(msg)
|
|
|
|
return cast(Callable[..., Awaitable[T]], wrapper)
|
|
|
|
return decorator
|