代码之家 › 专栏 › 技术社区 › Masoud Rahimi Chris

在python中以静态方式从文件导入类方法

apscheduler python

0

Masoud Rahimi Chris · 技术社区 · 8 年前

实际上,我找不到合适的话题。我想用 Apscheduler 图书馆。我为这个库创建了一个类方法,并自己添加了一些缺少的特性作为一个全新的调度程序库。当使用 SQLAlchemyJobStore 添加到调度程序的函数必须可序列化,但当我创建新类时,添加的函数是类本身的成员。例如,假设以下代码:

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)

回溯(最近的最后一次调用):文件 “..\python36\lib\site packages\apscheduler\executors\base.py”,行 125,运行中的作业 retval=job.func(*job.args,**job.kwargs)类型错误:tick()缺少1个必需的位置参数:“j”

此错误表示无法序列化整个类,但如果我将实例命名为与类名相同的名称(此处 sched )它会被修好的。在这个例子中, tick 功能必须成为日程班的一员,因为在我原来的班里 scheduler.add_job 从类方法获取已修改的函数,目前,我不想完全编辑新的调度类。

我的问题是,当我想从一个文件中导入类时,会发生确切的问题。有没有其他方法可以从文件中导入类?

2 回复 | 直到 8 年前

1

Alex Grönholm 8 年前

这个问题可以通过以下方法解决:

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

但是,这需要将调度程序从 sched 类,因为它不能与 计划 类

2

0

Masoud Rahimi Chris 8 年前

因为全班同学 sched 无法在数据库中序列化,类必须定义为静态。所以我把类改成了静态类。

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)