melobot.utils¶

melobot.utils.get_obj_name(obj: Any, otype: Literal['callable', 'class', 'object'] | str = 'object', default: str = '<anonymous %s>') → str[源代码]¶

获取一个对象的限定名称或名称，这适用于一些类型较宽的参数。

无法获取有效名称时，产生一个 default % otype 字符串

例如某处接受一个 Callable 类型的参数，对于一般函数来说，使用 __qualname__ 或 __name__ 可获得名称，但某些可调用对象这些值可能为 None 或不存在。使用此方法可保证一定返回字符串

def _(a: Callable) -> None:
    valid_str: str = get_obj_name(a, otype="callable")

def _(a: type) -> None:
    valid_str: str = get_obj_name(a, otype="class")

def _(a: Any) -> None:
    valid_str: str = get_obj_name(a, otype="type of a, only for str concat")

参数:

obj (Any) -- 对象
otype (Literal['callable', 'class', 'object'] | str) -- 预期的对象类型
default (str) -- 无法获取任何有效名称时的默认字符串

返回:

对象名称或默认字符串

返回类型:

str

melobot.utils.singleton(cls: Callable[[P], T]) → Callable[[P], T][源代码]¶

单例装饰器

参数:: cls (Callable[[~P], T]) -- 需要被单例化的可调用对象
返回:: 需要被单例化的可调用对象
返回类型:: Callable[[~P], T]

class melobot.utils.RWContext[源代码]¶

基类：object

异步读写上下文

提供异步安全的读写上下文。在读取时可以多读，同时读写互斥。

使用方法：

rwc = RWContext()
# 读时使用此控制器的安全读上下文：
async with rwc.read():
    ...
# 写时使用此控制器的安全写上下文：
async with rwc.write():
    ...

__init__(read_limit: int | None = None) → None[源代码]¶

初始化异步读写上下文

参数:: read_limit (int | None) -- 读取的数量限制，为空则不限制
返回类型:: None

read() → AsyncGenerator[None, None][源代码]¶

上下文管理器，展开一个关于该对象的安全异步读上下文

返回类型:: AsyncGenerator[None, None]

write() → AsyncGenerator[None, None][源代码]¶

上下文管理器，展开一个关于该对象的安全异步写上下文

返回类型:: AsyncGenerator[None, None]

melobot.utils.get_id() → str[源代码]¶

从 melobot 内部 id 获取器获得一个 id 值，不保证线程安全。算法使用雪花算法

返回:: id 值
返回类型:: str

melobot.utils.to_async(obj: Callable[[P], T] | AsyncCallable[P, T] | Awaitable[T]) → Callable[[P], Coroutine[Any, Any, T]][源代码]¶

异步包装函数

将一个可调用对象或可等待对象装饰为异步函数

参数:: obj (Callable[[~P], T] | AsyncCallable[~P, T] | Awaitable[T]) -- 需要转换的可调用对象或可等待对象
返回:: 异步函数
返回类型:: Callable[[~P], Coroutine[Any, Any, T]]

melobot.utils.to_coro(obj: Callable[[P], T] | AsyncCallable[P, T] | Awaitable[T], *args: Any, **kwargs: Any) → Coroutine[Any, Any, T][源代码]¶

协程包装函数

将一个可调用对象或可等待对象装饰为异步函数，并返回对应的协程

参数:

obj (Callable[[~P], T] | AsyncCallable[~P, T] | Awaitable[T]) -- 需要包装的可调用对象或可等待对象
args (Any) -- 需要使用的位置参数
kwargs (Any) -- 需要使用的关键字参数

返回:

协程

返回类型:

Coroutine[Any, Any, T]

melobot.utils.if_not(condition: Callable[[], Any] | AsyncCallable[(), Any] | bool, reject: AsyncCallable[(), None], give_up: bool = False) → Callable[[AsyncCallable[P, T]], AsyncCallable[P, T | None]][源代码]¶

条件判断装饰器

参数:

condition (Callable[[], Any] | AsyncCallable[(), Any] | bool) -- 用于判断的条件（如果是可调用对象，则先求值再转为 bool 值）
reject (AsyncCallable[(), None]) -- 当条件为 False 时，执行的回调
give_up (bool) -- 在条件为 False 时，是否放弃执行被装饰函数

返回类型:

Callable[[AsyncCallable[~P, T]], AsyncCallable[~P, T | None]]

melobot.utils.unfold_ctx(getter: Callable[[], ContextManager | AsyncContextManager]) → Callable[[AsyncCallable[P, T]], AsyncCallable[P, T]][源代码]¶

上下文装饰器

展开一个上下文，供被装饰函数使用。但注意此装饰器不支持获取上下文管理器 yield 的值

参数:: getter (Callable[[], ContextManager | AsyncContextManager]) -- 上下文管理器获取方法
返回类型:: Callable[[AsyncCallable[~P, T]], AsyncCallable[~P, T]]

melobot.utils.lock(callback: AsyncCallable[(), CbRetT] | None = None) → Callable[[AsyncCallable[P, OriginRetT]], AsyncCallable[P, CbRetT | OriginRetT]][源代码]¶

锁装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数加锁。

在获取锁冲突时，调用 callback 获得一个回调并执行。回调执行完毕后直接返回。

callback 参数为空，只应用 asyncio.Lock 的锁功能。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:: callback (AsyncCallable[(), CbRetT] | None) -- 获取锁冲突时的回调
返回类型:: Callable[[AsyncCallable[~P, OriginRetT]], AsyncCallable[~P, CbRetT | OriginRetT]]

melobot.utils.cooldown(busy_callback: ~melobot.typ.AsyncCallable[(), ~melobot.utils.FirstCbRetT] | None = None, cd_callback: ~melobot.typ.AsyncCallable[(<class 'float'>, ), ~melobot.utils.SecondCbRetT] | None = None, interval: float = 5) → Callable[[AsyncCallable[P, OriginRetT]], AsyncCallable[P, OriginRetT | FirstCbRetT | SecondCbRetT]][源代码]¶

冷却装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加 cd 时间。

如果被装饰函数已有一个在运行，此时调用 busy_callback 生成回调并执行。回调执行完毕后直接返回。

busy_callback 参数为空，则等待已运行的运行完成。随后执行下面的“冷却”处理逻辑。

当被装饰函数没有在运行的，但冷却时间未结束：

cd_callback 不为空：使用 cd_callback 生成回调并执行。
cd_callback 为空，被装饰函数持续等待，直至冷却结束再执行。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

busy_callback (AsyncCallable[(), FirstCbRetT] | None) -- 已运行时的回调
cd_callback (AsyncCallable[(<class 'float'>,), SecondCbRetT] | None) -- 冷却时间未结束的回调
interval (float) -- 冷却时间

返回类型:

Callable[[AsyncCallable[~P, OriginRetT]], AsyncCallable[~P, OriginRetT | FirstCbRetT | SecondCbRetT]]

melobot.utils.semaphore(callback: AsyncCallable[(), CbRetT] | None = None, value: int = -1) → Callable[[AsyncCallable[P, OriginRetT]], AsyncCallable[P, CbRetT | OriginRetT]][源代码]¶

信号量装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加信号量控制。

在信号量无法立刻获取时，将调用 callback 获得回调并执行。回调执行完毕后直接返回。

callback 参数为空，只应用 asyncio.Semaphore 的信号量功能。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

callback (AsyncCallable[(), CbRetT] | None) -- 信号量无法立即获取的回调
value (int) -- 信号量阈值

返回类型:

Callable[[AsyncCallable[~P, OriginRetT]], AsyncCallable[~P, CbRetT | OriginRetT]]

melobot.utils.timelimit(callback: AsyncCallable[(), CbRetT] | None = None, timeout: float = 5) → Callable[[AsyncCallable[P, OriginRetT]], AsyncCallable[P, CbRetT | OriginRetT]][源代码]¶

时间限制装饰器

本方法作为异步函数的装饰器使用，可以为被装饰函数添加超时控制。

超时之后，调用 callback 获得回调并执行，同时取消原任务。

callback 参数为空，如果超时，则抛出 asyncio.TimeoutError 异常。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值

参数:

callback (AsyncCallable[(), CbRetT] | None) -- 超时时的回调
timeout (float) -- 超时时间

返回类型:

Callable[[AsyncCallable[~P, OriginRetT]], AsyncCallable[~P, CbRetT | OriginRetT]]

melobot.utils.speedlimit(callback: AsyncCallable[(), CbRetT] | None = None, limit: int = 60, duration: int = 60) → Callable[[AsyncCallable[P, OriginRetT]], AsyncCallable[P, CbRetT | OriginRetT]][源代码]¶

流量/速率限制装饰器（使用固定窗口算法）

本方法作为异步函数的装饰器使用，可以为被装饰函数添加流量控制：duration 秒内只允许 limit 次调用。

超出调用速率限制后，调用 callback 获得回调并执行，同时取消原任务。

callback 参数为空，等待直至满足速率控制要求再调用。

被装饰函数的返回值：被装饰函数被执行 -> 被装饰函数返回值；执行任何回调 -> 那个回调的返回值。

参数:

callback (AsyncCallable[(), CbRetT] | None) -- 超出速率限制时的回调
limit (int) -- duration 秒内允许调用多少次
duration (int) -- 时长区间

返回类型:

Callable[[AsyncCallable[~P, OriginRetT]], AsyncCallable[~P, CbRetT | OriginRetT]]

melobot.utils.call_later(callback: Callable[[], None], delay: float) → TimerHandle[源代码]¶

同步函数延迟调度

在指定的 delay 后调度一个 callback 执行。callback 应该是同步方法。

参数:

callback (Callable[[], None]) -- 同步函数
delay (float) -- 多长时间后调度

返回:

asyncio.TimerHandle 对象

返回类型:

TimerHandle

melobot.utils.call_at(callback: Callable[[], None], timestamp: float) → TimerHandle[源代码]¶

同步函数指定时间调度

在指定的时间戳调度一个 callback 执行。callback 应该是同步方法。timestamp <= 当前时刻回调立即执行

参数:

callback (Callable[[], None]) -- 同步函数
timestamp (float) -- 在什么时刻调度

返回:

asyncio.TimerHandle 对象

返回类型:

TimerHandle

melobot.utils.async_later(callback: Coroutine[Any, Any, T], delay: float) → Future[T][源代码]¶

异步函数延迟调度（可自主选择是否等待）

在指定的 delay 后调度一个 callback 执行。callback 是协程。

返回一个 asyncio.Future 对象，你可以选择等待或不等待。等待 asyncio.Future 即是等待 callback 的返回值。

注意：如果 callback 未完成就被取消，需要捕获 asyncio.CancelledError。

参数:

callback (Coroutine[Any, Any, T]) -- 异步函数（可有返回值）
delay (float) -- 多长时间后调度

返回:

asyncio.Future 对象

返回类型:

Future[T]

melobot.utils.async_at(callback: Coroutine[Any, Any, T], timestamp: float) → Future[T][源代码]¶

异步函数指定时间调度（可自主选择是否等待）

在指定的时间戳调度一个 callback 执行。callback 是协程。

返回一个 asyncio.Future 对象，你可以选择等待或不等待。等待 asyncio.Future 即是等待 callback 的返回值。

注意：如果 callback 未完成就被取消，需要捕获 asyncio.CancelledError。

参数:

callback (Coroutine[Any, Any, T]) -- 异步函数（可有返回值）
timestamp (float) -- 在什么时刻调度

返回:

asyncio.Future 对象

返回类型:

Future[T]

melobot.utils.async_interval(callback: Callable[[], Coroutine[Any, Any, None]], interval: float) → Task[None][源代码]¶

异步函数间隔调度（类似 JavaScript 的 setInterval）

每过时间间隔执行 callback 一次。callback 是返回协程的可调用对象（异步函数或 lambda 函数等）。

返回一个 asyncio.Task 对象，可使用该 task 取消调度过程。

参数:

callback (Callable[[], Coroutine[Any, Any, None]]) -- 异步函数
interval (float) -- 调度的间隔

返回:

asyncio.Task 对象

返回类型:

Task[None]