APScheduler搭配RedisJobStore时继承类调用抽象父类方法报错

在Telegram API机器人开发中使用APScheduler时，遇到异常场景：当使用RedisJobStore存储任务时，继承抽象父类的子类任务会调用父类的抽象方法而非自身实现，但手动触发该函数时完全正常。我已在GitHub提交issue#767。

复现代码 #

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import asyncio
import logging

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore


logging.basicConfig(filename='logfile.log', level=logging.INFO)
logging.getLogger('apscheduler').setLevel(logging.DEBUG)


class MyAbstractClass(ABC):
    @classmethod
    @abstractmethod
    def get_required_members(cls):
        raise NotImplementedError

    @classmethod
    async def initiate_all(cls):
        members = cls.get_required_members()
        logging.info(f'Got {members=}')
        ...


class MyImplementation(MyAbstractClass):
    @classmethod
    def get_required_members(cls):
        return ['Alex', 'Anna']

    @classmethod
    def append_to_scheduler(cls, scheduler: AsyncIOScheduler, run_date: datetime):
        return scheduler.add_job(
            func=cls.initiate_all,
            trigger='date',
            run_date=run_date
        )


redis_job_store = RedisJobStore()
scheduler = AsyncIOScheduler(jobstores={'default': redis_job_store}, logger=logging.getLogger())
scheduler.start()

run_date = datetime.now() + timedelta(seconds=5)
MyImplementation.append_to_scheduler(scheduler=scheduler, run_date=run_date)
asyncio.get_event_loop().run_forever()

报错信息 #

Job "MyAbstractClass.initiate_all (trigger: date[2023-08-04 19:01:03 MSK], next run at: 2023-08-04 19:01:03 MSK)" raised an exception
Traceback (most recent call last):
  File "/media/russich555/hdd/Programming/Freelance/YouDo/29.2pilot/venv/lib/python3.11/site-packages/apscheduler/executors/base_py3.py", line 30, in run_coroutine_job
    retval = await job.func(*job.args, **job.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/media/russich555/hdd/Programming/Freelance/YouDo/29.2pilot/mre/api.py", line 13, in initiate_all
    members = cls.get_required_members()
              ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/media/russich555/hdd/Programming/Freelance/YouDo/29.2pilot/mre/api.py", line 9, in get_required_members
    raise NotImplementedError
NotImplementedError

问题原因 #

RedisJobStore默认序列化逻辑中，对类方法的引用仅保存了方法名称及其定义的父类信息，未正确保留子类的上下文。当任务从Redis反序列化执行时，initiate_all方法的cls参数被解析为父类MyAbstractClass，而非实际的子类MyImplementation，因此触发了父类的抽象方法报错。

解决方案 #

方案1：改用实例方法替代类方法 #

将类方法改为实例方法，通过子类实例传入调度器，序列化时会保留实例的类信息：

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import asyncio
import logging

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore


logging.basicConfig(filename='logfile.log', level=logging.INFO)
logging.getLogger('apscheduler').setLevel(logging.DEBUG)


class MyAbstractClass(ABC):
    @abstractmethod
    def get_required_members(self):
        raise NotImplementedError

    async def initiate_all(self):
        members = self.get_required_members()
        logging.info(f'Got {members=}')
        ...


class MyImplementation(MyAbstractClass):
    def get_required_members(self):
        return ['Alex', 'Anna']

    @classmethod
    def append_to_scheduler(cls, scheduler: AsyncIOScheduler, run_date: datetime):
        instance = cls()
        return scheduler.add_job(
            func=instance.initiate_all,
            trigger='date',
            run_date=run_date
        )


redis_job_store = RedisJobStore()
scheduler = AsyncIOScheduler(jobstores={'default': redis_job_store}, logger=logging.getLogger())
scheduler.start()

run_date = datetime.now() + timedelta(seconds=5)
MyImplementation.append_to_scheduler(scheduler=scheduler, run_date=run_date)
asyncio.get_event_loop().run_forever()

方案2：用functools.partial绑定子类上下文 #

通过functools.partial明确将子类绑定到类方法，确保序列化时保留正确的类引用：

from functools import partial

# 修改MyImplementation类的append_to_scheduler方法
@classmethod
def append_to_scheduler(cls, scheduler: AsyncIOScheduler, run_date: datetime):
    return scheduler.add_job(
        func=partial(cls.initiate_all),
        trigger='date',
        run_date=run_date
    )

方案3：更换序列化器 #

使用支持完整类上下文序列化的dill替代默认pickle（需先安装pip install dill）：

import dill

redis_job_store = RedisJobStore(pickle=dill)

内容的提问来源于stack exchange，提问作者555Russich