pip install python-crontab

Crontab module for read and writing crontab files and accessing system cron automatically and simply using a direct API.

 

Field Name	Mandatory	Allowed Values	Allowed Special Characters
Minutes	Yes	0-59	* / , –
Hours	Yes	0-23	* / , –
Day of month	Yes	1-31	* / , –
Month	Yes	1-12 or JAN-DEC	* / , –
Day of week	Yes	0-6 or SUN-SAT	* / , –

 

Supported special cases allow crontab lines to not use fields. These are supported aliases:

Case	Meaning
@reboot	Every boot
@hourly	0 * * * *
@daily	0 0 * * *
@weekly	0 0 * * 0
@monthly	0 0 1 * *
@yearly	0 0 1 1 *
@annually	0 0 1 1 *
@midnight	0 0 * * *

How to Use the Module

Getting access to a crontab can happen in four ways:

from crontab import CronTab

system_cron = CronTab()
user_cron = CronTab('root')
file_cron = CronTab(tabfile='filename.tab')
mem_cron = CronTab(tab="""
  * * * * * command
""")

Creating a new job is as simple as:

job  = cron.new(command='/usr/bin/echo')

And setting the job’s time restrictions:

job.minute.during(5,50).every(5)
job.hour.every(4)
job.day.on(4, 5, 6)

job.dow.on('SUN')
job.month.during('APR', 'NOV')

Creating a job with a comment:

job = cron.new(command='/foo/bar',comment='SomeID')

Disabled or Enable Job:

job.enable()
job.enable(False)
False == job.is_enabled()

Validity Check:

True == job.is_valid()

Use a special syntax:

job.every_reboot()

Find an existing job by command:

list = cron.find_command('bar')

Find an existing job by comment:

list = cron.find_comment('ID or some text')

Set and get the comment for a job:

comment = job.meta(['New Comment for job'])

Clean a job of all rules:

job.clear()

Iterate through all jobs:

for job in cron:
    print job

Iterate through all lines:

for line in cron.lines:
    print line

Remove Items:

cron.remove( job )
cron.remove_all('echo')

Write CronTab back to system or filename:

cron.write()

Write CronTab to new filename:

cron.write( 'output.tab' )

Log Functionality

The log functionality will read a cron log backwards to find you last run instances of your crontab and cron jobs.

The crontab will limit returned entries to user the crontab is for.

cron = CronTab(user=’root’)

for d in cron.log:

print d[‘pid’] + ” – ” + d[‘date’]

Each job can return a log iterator too, these are filtered so you can see when the last execution was.

for d in cron.find_command(‘echo’)[0].log:

print d[‘pid’] + ” – ” + d[‘date’]

Schedule Functionality

If you have croniter python module installed, you will have access to a schedule on each job. For example if you want to know when a job will next run:

schedule = job.schedule(date_from=datetime.now())

This creates a schedule croniter based on the job from time specified. The default date_from is current date/time if not specified. Next we can get datetime of the next job:

datetime = schedule.get_next()

Or the previous:

datetime = schedule.get_prev()

The get methods work in the same way as default croniter, except that they will return datetime objects by default instead of floats. If you want the original functionality, pass float into method when calling:

datetime = schedule.get_current(float)

If you don’t have croniter module installed, you’ll get an ImportError when you first try using schedule function on your cron job object.

CronManager Example

import argparse
import os ,sys
import logging
from crontab import CronTab
"""
Task Scheduler
==========
This module manages periodic tasks using cron.
"""
class CronManager:

def __init__(self):
    self.cron = CronTab(user=True)

def add_minutely(self, name, user, command, environment=None):
"""
Add an hourly cron task
"""
    cron_job = self.cron.new(command=command, user=user)
    cron_job.minute.every(2)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

def add_hourly(self, name, user, command, environment=None):
"""
Add an hourly cron task
"""
    cron_job = self.cron.new(command=command, user=user)
    cron_job.minute.on(0)
    cron_job.hour.during(0,23)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

def add_daily(self, name, user, command, environment=None):
"""
Add a daily cron task
"""
    cron_job = self.cron.new(command=command, user=user)
    cron_job.minute.on(0)
    cron_job.hour.on(0)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True
def add_weekly(self, name, user, command, environment=None):
"""
Add a weekly cron task
"""
    cron_job = self.cron.new(command=command)
    cron_job.minute.on(0)
    cron_job.hour.on(0)
    cron_job.dow.on(1)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

def add_monthly(self, name, user, command, environment=None):
"""
Add a monthly cron task
"""
    cron_job = self.cron.new(command=command)
    cron_job.minute.on(0)
    cron_job.hour.on(0)
    cron_job.day.on(1)
    cron_job.month.during(1,12)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

def add_quarterly(self, name, user, command, environment=None):
"""
Add a quarterly cron task
"""
    cron_job = self.cron.new(command=command)
    cron_job.minute.on(0)
    cron_job.hour.on(0)
    cron_job.day.on(1)
    cron_job.month.on(3,6,9,12)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

def add_anually(self, name, user, command, environment=None):
"""
Add a yearly cron task
"""
    cron_job = self.cron.new(command=command)
    cron_job.minute.on(0)
    cron_job.hour.on(0)
    cron_job.month.on(12)
    cron_job.enable()
    self.cron.write()
    if self.cron.render():
        print self.cron.render()
        return True

 

Drop me a comment if you have any queries and will get back.