You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在APScheduler任务中使用异步类方法?解决协程未等待问题

APScheduler异步类方法任务不执行,报“协程从未被等待”警告

我开发机器人时使用APScheduler的AsyncIOScheduler和AsyncIOExecutor,普通异步测试函数(几秒后发送“pong”消息)能正常运行,但异步类方法完全没有执行,还会抛出RuntimeWarning: 协程从未被等待。

我的APS配置极简,用SQLite作为轻量任务持久层:

def new_scheduler():
    jobstores = {
        "memory": MemoryJobStore(),
        "default": SQLAlchemyJobStore(url="sqlite:///.jobs.sqlite"),
    }
    executors = {
        "default": AsyncIOExecutor(),
    }
    return AsyncIOScheduler(
        jobstores=jobstores, executors=executors, timezone=datetime.timezone.utc
    )

问题明显出在类方法的使用上——调度类外定义的异步任务运行正常,但类方法任务完全失效。

涉及的ModAction.lift是用户限制基类的类方法,负责移除限制并从数据库中删除记录。添加限制时如果设置了过期时间,我会给调度器添加一个调用lift的任务,同时存储任务ID,方便手动提前解除限制时取消任务:

class ModAction(Document): # Document继承自pydantic.BaseModel,提供数据库交互的核心字段和方法
    # FakeMember是序列化类,仅保留用户和服务器ID
    issuer: FakeMember
    target: FakeMember
    reason: Optional[str] = None
    expires: Optional[datetime.datetime] = None
    lift_job_id: Optional[str] = None

    async def apply(self, bot: Bot, no_dm: bool = False):
        if not no_dm:
            await try_dm(await self.target.get_or_fetch(bot), embeds=[self.dm_embed])
        if self.expires is not None and self.expires > now():
            self.lift_job_id = bot.scheduler.add_job(
                self.__class__.lift,
                kwargs={
                    "id": self.id,
                    "reason": "Expired",
                },
                trigger="date",
                next_run_time=self.expires,
            ).id
        await bot.log(embed=self.apply_log_embed(bot))
        await self.store(bot.db_connection)

    @classmethod
    @validate_call
    async def lift(
        cls,
        id: DocumentId,
        reason: Optional[str] = None,
        member: Optional[FakeMember] = None,
    ):
        from cogs.mod import _bot # 加载扩展的bot实例存储在这个全局变量中

        self = await cls.fetch(id, _bot.db_connection)
        await _bot.log(embed=self.lift_log_embed(_bot, reason, member))
        await _bot.db_connection.delete(id)

不同管理操作有不同执行逻辑,所以用类方法实现。以服务器封禁为例:

class ServerBan(ModAction):
    async def apply(self, bot: Bot, no_dm: bool = False):
        await super().apply(bot, no_dm)
        await self.target.get(bot).ban(reason=self.apply_audit_reason(bot))

    @classmethod
    @validate_call
    async def lift(
        cls,
        id: DocumentId,
        reason: Optional[str] = None,
        member: Optional[FakeMember] = None,
    ):
        from cogs.mod import _bot

        self: cls = await cls.fetch(id, _bot.db_connection)
        await super().lift(id, reason, member)
        await self.target.get(_bot).unban(
            reason=cls.lift_audit_reason(_bot, reason, member)
        )

想问下这是APScheduler的问题?类方法本身的问题?还是我的使用方式有误?
由于机器人仍处于早期开发阶段,我不执着于使用APScheduler,无论是修复当前问题的方法,还是全新的替代方案,都欢迎提出建议。

内容的提问来源于stack exchange,提问作者M1N3R

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.13 16:55:18