我正在寻找Python中的一个库,它将提供at和cron之类的功能。

我非常希望有一个纯Python的解决方案,而不是依赖于安装在盒子上的工具;这样我在没有cron的机器上运行。

对于那些不熟悉cron的人:你可以根据一个表达式来安排任务,比如:

 0 2 * * 7 /usr/bin/run-backup # run the backups at 0200 on Every Sunday
 0 9-17/2 * * 1-5 /usr/bin/purge-temps # run the purge temps command, every 2 hours between 9am and 5pm on Mondays to Fridays.

cron时间表达式语法不那么重要,但我希望具有这种灵活性。

如果没有什么东西可以为我开箱即用,任何关于构建模块的建议都将被感激地接受。

编辑 我对启动进程不感兴趣,只对同样用Python编写的“作业”——Python函数感兴趣。我认为这将是一个不同的线程,但不是在不同的进程中。

为此,我正在寻找cron time表达式的表达性,但在Python中。

Cron已经存在多年了,但我正在努力尽可能地实现可移植性。我不能指望它的存在。


当前回答

如果你正在寻找一个轻量级的结账时间表:

import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while 1:
    schedule.run_pending()
    time.sleep(1)

披露:我是这个库的作者。

其他回答

我不知道这样的东西是否已经存在。编写自己的time, datetime和/或日历模块很容易,请参阅http://docs.python.org/library/time.html

python解决方案的唯一问题是,您的作业需要始终运行,并且可能在重新启动后自动“复活”,为此您确实需要依赖于依赖于系统的解决方案。

我喜欢pycron包解决这个问题的方式。

import pycron
import time

while True:
    if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
        print('running backup')
        time.sleep(60)               # The process should take at least 60 sec
                                     # to avoid running twice in one minute
    else:
        time.sleep(15)               # Check again in 15 seconds

如果你正在寻找一个轻量级的结账时间表:

import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while 1:
    schedule.run_pending()
    time.sleep(1)

披露:我是这个库的作者。

列出的解决方案甚至都没有尝试解析复杂的cron调度字符串。这是我的版本,用croniter。基本要点:

schedule = "*/5 * * * *" # Run every five minutes

nextRunTime = getNextCronRunTime(schedule)
while True:
     roundedDownTime = roundDownTime()
     if (roundedDownTime == nextRunTime):
         ####################################
         ### Do your periodic thing here. ###
         ####################################
         nextRunTime = getNextCronRunTime(schedule)
     elif (roundedDownTime > nextRunTime):
         # We missed an execution. Error. Re initialize.
         nextRunTime = getNextCronRunTime(schedule)
     sleepTillTopOfNextMinute()

辅助例程:

from croniter import croniter
from datetime import datetime, timedelta

# Round time down to the top of the previous minute
def roundDownTime(dt=None, dateDelta=timedelta(minutes=1)):
    roundTo = dateDelta.total_seconds()
    if dt == None : dt = datetime.now()
    seconds = (dt - dt.min).seconds
    rounding = (seconds+roundTo/2) // roundTo * roundTo
    return dt + timedelta(0,rounding-seconds,-dt.microsecond)

# Get next run time from now, based on schedule specified by cron string
def getNextCronRunTime(schedule):
    return croniter(schedule, datetime.now()).get_next(datetime)

# Sleep till the top of the next minute
def sleepTillTopOfNextMinute():
    t = datetime.utcnow()
    sleeptime = 60 - (t.second + t.microsecond/1000000.0)
    time.sleep(sleeptime)

我知道有很多答案,但另一个解决方案可能是找装修师。这是一个每天在特定时间重复一个函数的例子。使用这种方式很酷的想法是,你只需要将Syntactic Sugar添加到你想要调度的函数中:

@repeatEveryDay(hour=6, minutes=30)
def sayHello(name):
    print(f"Hello {name}")

sayHello("Bob") # Now this function will be invoked every day at 6.30 a.m

装饰器看起来像这样:

def repeatEveryDay(hour, minutes=0, seconds=0):
    """
    Decorator that will run the decorated function everyday at that hour, minutes and seconds.
    :param hour: 0-24
    :param minutes: 0-60 (Optional)
    :param seconds: 0-60 (Optional)
    """
    def decoratorRepeat(func):

        @functools.wraps(func)
        def wrapperRepeat(*args, **kwargs):

            def getLocalTime():
                return datetime.datetime.fromtimestamp(time.mktime(time.localtime()))

            # Get the datetime of the first function call
            td = datetime.timedelta(seconds=15)
            if wrapperRepeat.nextSent == None:
                now = getLocalTime()
                wrapperRepeat.nextSent = datetime.datetime(now.year, now.month, now.day, hour, minutes, seconds)
                if wrapperRepeat.nextSent < now:
                    wrapperRepeat.nextSent += td

            # Waiting till next day
            while getLocalTime() < wrapperRepeat.nextSent:
                time.sleep(1)

            # Call the function
            func(*args, **kwargs)

            # Get the datetime of the next function call
            wrapperRepeat.nextSent += td
            wrapperRepeat(*args, **kwargs)

        wrapperRepeat.nextSent = None
        return wrapperRepeat

    return decoratorRepeat