python之从 python 中的类似 cron 的调度程序中删除事件

所以我创建了一个 cron-like-scheduler，正如在此处找到的问题的答案中所建议的:

How do I get a Cron like scheduler in Python?

但有时，如果一个事件发生，我想删除一个不同的事件....

我将如何创建此功能？....基本上，一旦创建，我如何删除事件？

另外....我在从事件中创建事件时遇到了问题。基本上，我的目标是每小时使用一个事件来解析一个文件，它包含我想做其他事件的时间......所以我想从这个每小时事件中创建新事件...... .但它们似乎在创建后就消失了....

谢谢!

请您参考如下方法:

这完全取决于您实际如何实现您的 cronjob 运行。但是，如果您希望事件能够修改其他事件，我可以想到 3 个选项:

1。将对 cron 的引用传递给函数

您的 CronTab 类必须提供一种方法让事件能够“检查”它。最好的选择是使用“cron”参数扩展关键字参数并将其传递给函数，如下所示:

class CronTab(object): 
    def __init__(self, *events): 
        self.events = events 
        for event in self.events: 
             event.kwargs['cron'] = self 
# rest of the calss remains unmodified 

这样一来，所有从运行的 Crontab 调用的函数都将传递一个“cron”参数，并引用运行它的 cron

2。添加“预运行”编码(marshal)

您扩展您的 CronTab 类，不仅可以按特定时间表执行事件，还可以在这些事件发生之前调用其他函数。获取事件实例并且必须返回 True 才能运行的函数。如果它们中的任何一个返回其他内容，则相关事件将不会运行。

一个基本的(未经测试的)实现是:

class CronTab(object): 
    def __init__(self, *events, marshalls=None): 
        self.events = events 
        if marshalls is not None: 
            # marshalls must be a list of callables that return True or False 
            self.marshalls = marshalls 
        else 
            self.marshalls = [] 
 
    def run(self): 
        t=datetime(*datetime.now().timetuple()[:5]) 
        while 1: 
            for e in self.events: 
                if all([x(e) for x in self.marshalls]): 
                    e.check(t) 
 
            t += timedelta(minutes=1) 
            while datetime.now() < t: 
                time.sleep((t - datetime.now()).seconds) 

显然有更好的方法来实现这一点。例如，通过执行 reduce 操作，从而避免在其中一个事件已返回 False 后为所有事件运行所有编码。但这应该为您指明正确的方向。

如果 CronTab 类实际上做了比调用事件的实例 check 更多的工作，那么实现这个也可能更容易 - 或者至少将运行检查与实际运行事件分离 -但我不想对原始代码进行太多修改。

3。提供事件 Hook

这基本上与之前的解决方案相同，但更“细粒度”。您将拥有多个列表，而不是只有一个预运行编码列表。您可以根据自己的需要定制确切的数量和详细信息，但这里是我根据您的要求采取的做法。

除了一般的“预运行编码”之外，您还可以为每个事件名称创建一个这样的列表。在运行一个事件之前，您不仅要运行所有通用的“运行前编码”，还要运行特定于该事件的那些。

一种实现方法是将“预运行编码”列表制作成列表字典，并提供一些方法来管理此类列表。像这样:

class CronTab(object): 
    def __init__(self, *events, marshalls=None): 
        self.events = events 
        self.marshalls = dict() 
        if marshalls is not None: 
            # marshalls must be a list of callables that return True or False 
            self.marshalls['__general'] = marshalls 
        else 
            self.marshalls['__general'] = [] 
 
    # event is a string with the name of the event you want to marshall 
    # or None if you want to marshall all of them  
    def register(self, callback, event=None): 
        if event = None: 
            self.marshalls['__general'].append(callback) 
        else: 
            self.marshalls.setdefault(event, list()).append(callback) 
 
    def run(self): 
        t=datetime(*datetime.now().timetuple()[:5]) 
        while 1: 
            for e in self.events: 
                if all([x(e) for x in self.marshalls['__general']]) and 
                   all([x(e) for x in self.marshalls[e.name]]): 
                    # the above assumes e.name exists (not in reference implementation) 
                    e.check(t) 
 
        t += timedelta(minutes=1) 
        while datetime.now() < t: 
            time.sleep((t - datetime.now()).seconds) 

以同样的方式，您可以添加运行后 Hook 和其他您认为合适的奇怪东西。

---

关于您关于事件消失的问题，请向我们展示一些代码以便能够进一步解决问题。