如果你不知道Crontab是什么，它是一個用于Unix和類Unix操作系統(tǒng)的實用工具，允許你在特定時間自動安排和執(zhí)行任務(wù)。Crontab的一個很大優(yōu)勢是，它在崩潰或重啟后仍然有效。Crontab條目保存在由cron守護(hù)進(jìn)程讀取的文件中，該守護(hù)進(jìn)程在系統(tǒng)啟動時自動啟動。

本文的目的是使用python-crontab庫編寫Python程序的執(zhí)行計劃。這個庫允許你使用API在操作系統(tǒng)中生成crontab文件，而不必使用特定的操作系統(tǒng)命令。

首先，我們安裝該庫：

poetry add python-crontab

然后，我們創(chuàng)建一個名為create_crontab.py的文件，并包含以下內(nèi)容：

if __name__ == '__main__':
    cron_tab = CronTab(user=True)

    list_cron(cron_tab)
    delete_cron(cron_tab)
    create_cron(cron_tab)
    list_cron(cron_tab)

• cron_tab = CronTab(user=True): 我們創(chuàng)建一個與當(dāng)前用戶crontab關(guān)聯(lián)的CronTab對象。這將允許我們訪問和操作計劃任務(wù)。
• list_cron(cron_tab): 我們將創(chuàng)建一個名為list_cron的函數(shù)，用于在屏幕上顯示所有計劃任務(wù)的列表。 
• delete_cron(cron_tab): 我們將創(chuàng)建一個函數(shù)，用于在創(chuàng)建計劃任務(wù)之前從crontab中刪除所有計劃任務(wù)。這一步在更新計劃定義時非常有用。 
• create_cron(cron_tab): 我們將實現(xiàn)一個create_cron函數(shù)，用于創(chuàng)建新的計劃任務(wù)并將它們添加到crontab中。

列出計劃任務(wù)

我們將CronTab對象傳遞給此函數(shù)并迭代不同的任務(wù)。

def list_cron(cron):
 for job in cron:
     print(job)

刪除計劃任務(wù)

在進(jìn)行更改時，總是強(qiáng)制crontab寫入。

def delete_cron(cron):
 cron.remove_all()
 cron.write()

創(chuàng)建計劃任務(wù)

假設(shè)我們要安排執(zhí)行位于以下絕對路徑的程序：

/Users/xavierescudero/Projects/tutorial-trading-bot/tutorial_trading_bot/exchange_historical_importer.py

Crontab腳本在后臺運行，不在我們的項目內(nèi)，因此它不知道模塊路徑。我們需要給它指示：

• 進(jìn)入根目錄：cd /Users/xavierescudero/Projects/tutorial-trading-bot
• 在Poetry管理的虛擬環(huán)境中運行模塊：poetry run -m tutorial_trading_bot.exchange_historical.importer
• 使用-m選項，Python會在系統(tǒng)搜索目錄和當(dāng)前目錄中查找模塊，導(dǎo)入它，并像獨立腳本文件一樣運行它。

模塊目錄的位置

我們可以使用pathlib從create_crontab.py模塊文件的相對路徑獲取項目的根目錄：

from pathlib import Path
PROJECT_DIR_PATH = str(Path(__file__).parent.parent)

并創(chuàng)建到此目錄的目錄更改命令鏈：

CD_PROJECT_DIR_COMMAND = ''.join(['cd ', PROJECT_DIR_PATH])

定義要執(zhí)行的命令

我們創(chuàng)建執(zhí)行的初始部分，這將用于任何模塊：

EXECUTOR_COMMAND = ' '.join(['poetry', 'run python -m tutorial_trading_bot.'])

我們構(gòu)建了運行每個模塊的完整命令（包括目錄更改）：

EXECUTOR_PATH = ';'.join([CD_PROJECT_DIR_COMMAND, EXECUTOR_COMMAND])

HISTORICAL_IMPORTER_JOB = ''.join([EXECUTOR_PATH, 'exchange_historical_importer', ' &'])
TRADING_BOT_JOB = ''.join([EXECUTOR_PATH, 'trading_bot', ' &'])

定義計劃任務(wù)

現(xiàn)在我們有了命令，使用python-crontab庫的new()函數(shù)創(chuàng)建計劃任務(wù)，并用setall設(shè)置cron表達(dá)式。

def create_cron(cron):
 """
 Check crontab values using https://crontab.guru
 """
 cron.new(command=HISTORICAL_IMPORTER_JOB).setall('59 21 * * *') # At 21:59 every day
 cron.new(command=TRADING_BOT_JOB).setall('59 22 * * *')
 cron.write()

在網(wǎng)站https://crontab.guru上，你可以驗證你的cron表達(dá)式，并找到大量示例。

重啟時執(zhí)行（無需計劃）

當(dāng)你不在家時，發(fā)現(xiàn)Python進(jìn)程在重啟后沒有重新啟動是一件糟糕的事。這可能很令人沮喪，因為這意味著你的自動化任務(wù)或服務(wù)沒有運行。

我們還可以使用python-crontab庫在重啟時啟動程序：

cron.new(command=TRADING_BOT_JOB).every_reboot()

設(shè)置cron任務(wù)

我們現(xiàn)在可以從shell安裝我們的編程：

poetry run python -m tutorial-trading-bot.create_crontab

我們將在屏幕上看到計劃任務(wù)的列表：

2024-06-07 00:15:46,872 - __main__ - INFO - Crontab configured
59 21 * * *