从 python 中的静态文件导入 class 方法

Question

其实我找不到合适的话题。我正在尝试使用 Apscheduler 库。我为库创建了一个 class 方法，并自己添加了一些缺失的功能作为一个全新的调度程序库。当使用 SQLAlchemyJobStore 时，添加到调度程序 的函数必须 是可序列化的，但是当我创建一个新的 class 时，添加的函数是 class 本身的成员.例如，假设此代码：

from datetime import datetime
from time import sleep
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class sched:
    def __init__(self):
        jobstores = {
            'default': SQLAlchemyJobStore(url='sqlite:///test.db')
        }
        self.scheduler = BackgroundScheduler(jobstores=jobstores)

    def add_func(self, some_func, args={}, kwargs={}):
        self.scheduler.add_job(
            some_func, 'interval', args=args, kwargs=kwargs, seconds=5, replace_existing=True)

    def start(self):
        self.scheduler.start()


    def tick(self, i, j, **kwargs):
        print("hello{} {}".format(i, j))
        for k in kwargs:
            print(k, ":", kwargs[k])

if __name__ == '__main__':
    schedule = sched()
    schedule.start()
    schedule.add_func(schedule.tick, args=[1, 2], kwargs=dict(a=1, b=2))
    # using the following three lines solve the problem
    # sched = sched()
    # sched.start()
    # sched.add_func(sched.tick, args=[1, 2], kwargs=dict(a=1, b=2))
    while True:
        sleep(.1)

Traceback (most recent call last): File "...\Python36\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job retval = job.func(*job.args, **job.kwargs) TypeError: tick() missing 1 required positional argument: 'j'

此错误意味着整个 class 无法序列化，但如果我将我的实例命名为 class 名称（此处为 sched），它将得到修复。在示例中，tick 函数必须是时间表 class 本身的成员，因为在我原来的 class scheduler.add_job 中得到了修改class 方法的功能，目前，我不想完全编辑我的新时间表 class。

我的问题是，当我想从文件中导入 class 时，就会出现确切的问题。有没有其他方法可以从文件中导入 class？

Answer 1

问题可以通过以下方式规避：

schedule.add_func(sched.tick, args=[schedule, 1, 2], kwargs=dict(a=1, b=2))

但是，这需要您将调度程序移出 sched class，因为它不能与 sched class 的实例一起序列化.

Answer 2

因为整个class sched不能在数据库中序列化，所以class必须定义为静态的。所以我将 class 更改为静态 class.

from datetime import datetime
from time import sleep
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class sched:
    jobstores = {
        'default': SQLAlchemyJobStore(url='sqlite:///test.db')
    }
    scheduler = BackgroundScheduler(jobstores=jobstores)
    @classmethod
    def add_func(cls, some_func, args={}, kwargs={}):
        cls.scheduler.add_job(
            some_func, 'interval', args=args, kwargs=kwargs, seconds=5, replace_existing=True)

    @classmethod
    def start(cls):
        cls.scheduler.start()

    @classmethod
    def tick(cls, i, j, **kwargs):
        print("hello{} {}".format(i, j))
        for k in kwargs:
            print(k, ":", kwargs[k])

if __name__ == '__main__':
    schedule = sched()
    schedule.start()
    schedule.add_func(schedule.tick, args=[1, 2], kwargs=dict(a=1, b=2))
    while True:
        sleep(.1)

从 python 中的静态文件导入 class 方法

import a class method from file as static in python

python

apscheduler