任務(wù)自動(dòng)化 :用 Python 創(chuàng)建 Crontab 任務(wù)
如果你不知道Crontab是什么,它是一個(gè)用于Unix和類(lèi)Unix操作系統(tǒng)的實(shí)用工具,允許你在特定時(shí)間自動(dòng)安排和執(zhí)行任務(wù)。Crontab的一個(gè)很大優(yōu)勢(shì)是,它在崩潰或重啟后仍然有效。Crontab條目保存在由cron守護(hù)進(jìn)程讀取的文件中,該守護(hù)進(jìn)程在系統(tǒng)啟動(dòng)時(shí)自動(dòng)啟動(dòng)。
本文的目的是使用python-crontab庫(kù)編寫(xiě)Python程序的執(zhí)行計(jì)劃。這個(gè)庫(kù)允許你使用API在操作系統(tǒng)中生成crontab文件,而不必使用特定的操作系統(tǒng)命令。
首先,我們安裝該庫(kù):
poetry add python-crontab
然后,我們創(chuàng)建一個(gè)名為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)建一個(gè)與當(dāng)前用戶(hù)crontab關(guān)聯(lián)的CronTab對(duì)象。這將允許我們?cè)L問(wèn)和操作計(jì)劃任務(wù)。
- list_cron(cron_tab): 我們將創(chuàng)建一個(gè)名為list_cron的函數(shù),用于在屏幕上顯示所有計(jì)劃任務(wù)的列表。
- delete_cron(cron_tab): 我們將創(chuàng)建一個(gè)函數(shù),用于在創(chuàng)建計(jì)劃任務(wù)之前從crontab中刪除所有計(jì)劃任務(wù)。這一步在更新計(jì)劃定義時(shí)非常有用。
- create_cron(cron_tab): 我們將實(shí)現(xiàn)一個(gè)create_cron函數(shù),用于創(chuàng)建新的計(jì)劃任務(wù)并將它們添加到crontab中。
列出計(jì)劃任務(wù)
我們將CronTab對(duì)象傳遞給此函數(shù)并迭代不同的任務(wù)。
def list_cron(cron):
for job in cron:
print(job)
刪除計(jì)劃任務(wù)
在進(jìn)行更改時(shí),總是強(qiáng)制crontab寫(xiě)入。
def delete_cron(cron):
cron.remove_all()
cron.write()
創(chuàng)建計(jì)劃任務(wù)
假設(shè)我們要安排執(zhí)行位于以下絕對(duì)路徑的程序:
/Users/xavierescudero/Projects/tutorial-trading-bot/tutorial_trading_bot/exchange_historical_importer.py
Crontab腳本在后臺(tái)運(yùn)行,不在我們的項(xiàng)目?jī)?nèi),因此它不知道模塊路徑。我們需要給它指示:
- 進(jìn)入根目錄:cd /Users/xavierescudero/Projects/tutorial-trading-bot
- 在Poetry管理的虛擬環(huán)境中運(yùn)行模塊:poetry run -m tutorial_trading_bot.exchange_historical.importer
- 使用-m選項(xiàng),Python會(huì)在系統(tǒng)搜索目錄和當(dāng)前目錄中查找模塊,導(dǎo)入它,并像獨(dú)立腳本文件一樣運(yùn)行它。
模塊目錄的位置
我們可以使用pathlib從create_crontab.py模塊文件的相對(duì)路徑獲取項(xiàng)目的根目錄:
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)建了運(yùn)行每個(gè)模塊的完整命令(包括目錄更改):
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', ' &'])
定義計(jì)劃任務(wù)
現(xiàn)在我們有了命令,使用python-crontab庫(kù)的new()函數(shù)創(chuàng)建計(jì)劃任務(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上,你可以驗(yàn)證你的cron表達(dá)式,并找到大量示例。
重啟時(shí)執(zhí)行(無(wú)需計(jì)劃)
當(dāng)你不在家時(shí),發(fā)現(xiàn)Python進(jìn)程在重啟后沒(méi)有重新啟動(dòng)是一件糟糕的事。這可能很令人沮喪,因?yàn)檫@意味著你的自動(dòng)化任務(wù)或服務(wù)沒(méi)有運(yùn)行。
我們還可以使用python-crontab庫(kù)在重啟時(shí)啟動(dòng)程序:
cron.new(command=TRADING_BOT_JOB).every_reboot()
設(shè)置cron任務(wù)
我們現(xiàn)在可以從shell安裝我們的編程:
poetry run python -m tutorial-trading-bot.create_crontab
我們將在屏幕上看到計(jì)劃任務(wù)的列表:
2024-06-07 00:15:46,872 - __main__ - INFO - Crontab configured
59 21 * * *