Review request: a simple Cron wrapper in Python

Question

This is my first attempt to create a reusable and tested module. Any comments is highly appreciated.

#-*- coding: utf-8 -*-
"""
Micron: a micro wrapper around Unix crontab.
Micron is a thin wrapper around Unix crontab command. It let you add and remove
jobs from the crontab file or remove the entire crontab file. "crontab" is the
program used to install, deinstall or list the tables used to drive the cron
daemon
How it works:
>>> #Obviously you need to import the module
>>> import micron
>>> #Then you create a crontab instance
>>> crontab = micron.CronTab()
>>> #Assuming your crontab does not exist trying to read it will rise an error
>>> crontab.read()
Traceback (most recent call last):
 ...
 raise CronTabMissing()
CronTabMissing: Current crontab does not exist
>>> #Now you can add some jobs. Micron supports some common presets
>>> sorted(crontab.PRESETS.items())
[('daily', '0 0 * * * '), ('hourly', '0 * * * * '), ('monthly', '0 0 1 * * '), ('weekly', '0 0 * * 1 ')]
>>> crontab.add_job('weekly', 'echo "WOW"', 0)
>>> crontab.add_job('daily', 'echo "BOOM!"', 1)
>>> #You can omit the job id and micron will generate it for you
>>> crontab.add_job('daily', 'echo "BOOM!"')
>>> #Read again the crontab content
>>> crontab.read()
['0 0 * * 1 echo "WOW" #MICRON_ID_0', '0 0 * * * echo "BOOM!" #MICRON_ID_1', '0 0 * * * echo "BOOM!" #MICRON_ID_2']
>>> #See how it look in the crontab file
>>> print crontab
0 0 * * 1 echo "WOW" #MICRON_ID_0
0 0 * * * echo "BOOM!" #MICRON_ID_1
0 0 * * * echo "BOOM!" #MICRON_ID_2
>>> #Remove job with id 0
>>> crontab.remove_job(0)
>>> print crontab
0 0 * * * echo "BOOM!" #MICRON_ID_1
0 0 * * * echo "BOOM!" #MICRON_ID_2
>>> #Remove job with non existing id and you'll get and error
>>> crontab.remove_job(11)
Traceback (most recent call last):
 ...
 raise ValueError('Id %s not in crontab' % job_id)
ValueError: Id 11 not in crontab
>>> #If the presets are not enough, you can add your own timing using the
>>> #crontab syntax
>>> crontab.add_job('* * * * * ', 'echo "This will work"')
>>> #But you must use the correct syntax
>>> crontab.add_job('*wrong syntax* ', 'echo "This will not work"')
Traceback (most recent call last):
 ...
 raise CronTabSyntaxError(added_job[0])
CronTabSyntaxError: Syntax error adding:
 *wrong syntax* echo "This will not work" #MICRON_ID_4
<BLANKLINE>
>>> #Sometimes you want to remove all the jobs added by Micron
>>> crontab.remove_all_jobs()
>>> #Any other crontab content is not removed
>>> crontab.read()
[]
>>> #Remove the entire crontab file
>>> crontab.remove_crontab()
"""
from subprocess import Popen, PIPE, CalledProcessError
class CronTabError(Exception):
 """Base error class."""
 pass
class CronTabMissing(CronTabError):
 """Cron tab missing."""
 def __str__(self):
 return "Current crontab does not exist"
class CronTabSyntaxError(CronTabError):
 """Crontab syntax error."""
 def __init__(self, string):
 self.string = string
 def __str__(self):
 return "Syntax error adding:\n %s" % self.string
class CronTabIdError(CronTabError):
 """Crontab job already exists."""
 def __init__(self, cron_id):
 self.cron_id = cron_id
 def __str__(self):
 return "Job ID %s already exists" % self.cron_id
class CronTab():
 """A simple UNIX crontab wrapper
 This class let you interact with the crontab file of the current user
 reading, adding and removing lines.
 """
 CRONTAB_PATH = "/usr/bin/crontab"
 PRESETS = {
 "hourly":"0 * * * * ",
 "daily": "0 0 * * * ",
 "weekly":"0 0 * * 1 ",
 "monthly":"0 0 1 * * ",
 }
 PLACEHOLDER = ' #MICRON_ID_' #must start with a space, must be separated by "_"
 def __str__(self):
 jobs = self.read()
 return "\n".join(jobs)
 def read(self):
 """Read the crontab file content
 Return the content as a list of lines.
 """
 args = [self.CRONTAB_PATH, '-l']
 process = Popen(args, stdout=PIPE, stderr=PIPE)
 output, unused_err = process.communicate()
 retcode = process.poll()
 if retcode == 1:
 raise CronTabMissing()
 elif retcode:
 raise CalledProcessError(retcode, self.CRONTAB_PATH)
 jobs = output.splitlines()
 return jobs
 def _get_id(self, job):
 job_id = job.split(self.PLACEHOLDER)[-1]
 return int(job_id)
 def _create_id(self):
 """Create a unique id for a cron job."""
 try:
 jobs = self.read()
 job_ids = sorted([self._get_id(job) for job in jobs if self.PLACEHOLDER in job])
 if job_ids:
 new_job_id = int(job_ids[-1]) + 1
 else:
 new_job_id = 0
 except CronTabMissing:
 new_job_id = 0
 return new_job_id
 def save(self, job_list):
 """Overwrite the current crontab."""
 process = Popen([self.CRONTAB_PATH, '-'], stdout=PIPE, stdin=PIPE, stderr=PIPE)
 content = "\n".join(job_list)
 process.communicate(input=content)
 if process.returncode != 0:
 current_jobs = self.read()
 added_job = list(set(job_list) - set(current_jobs))
 raise CronTabSyntaxError(added_job[0])
 def add_job(self, timing, program, job_id=None):
 """Add a job to the current user crontab."""
 if self.PRESETS.has_key(timing):
 new_job = self.PRESETS[timing] + program 
 else:
 new_job = timing + program
 if job_id is None:
 job_id = self._create_id()
 new_job += "%s%s\n" % (self.PLACEHOLDER, job_id)
 try:
 jobs = self.read()
 except CronTabMissing:
 jobs = []
 for job in jobs:
 if self.PLACEHOLDER in job:
 current_job_id = self._get_id(job)
 if current_job_id == job_id:
 raise CronTabIdError(job_id)
 jobs.append(new_job)
 self.save(jobs)
 def remove_job(self, job_id):
 """Delete a job with the given job id."""
 jobs = self.read()
 for job in jobs:
 if self.PLACEHOLDER in job:
 current_job_id = self._get_id(job)
 if current_job_id == job_id:
 jobs.remove(job)
 self.save(jobs)
 break
 else:
 raise ValueError('Id %s not in crontab' % job_id)
 def remove_all_jobs(self):
 """Delete all jobs created by micron."""
 jobs = self.read()
 other_jobs = [job for job in jobs if self.PLACEHOLDER not in job]
 self.save(other_jobs)
 def remove_crontab(self):
 """Remove the crontab file.
 Remove the crontab file of the current user. The information
 contained in the crontab file is permanently lost.
 """
 process = Popen([self.CRONTAB_PATH, '-r'], stdout=PIPE, stderr=PIPE)
 output, unused_err = process.communicate()
 if process.returncode:
 raise CronTabError
if __name__ == "__main__":
 import doctest
 doctest.testmod()

Answer 1

I think it is a pretty good start:

• I would like to see a Job class.
• The stderr output would be great in cases of an retcode != 0.
• A validation and/or escaping of timing and program would be great.

• Raben: Thanks for the feedback. What do you mean with: "The stderr output would be great in cases of an retcode != 0."? You think I should include the stderr output when I raise one of the custom exception? It seems a good idea.
• dmeister: Yes, that is the idea.

Answer 2

I concur with dmeister.

Jobs are treated too opaquely by Micron and therefore have no access. A crontab entry is highly structured and a good Job class would allow inspection and modification of its elements.

This leads me to think that you have focused on the container (a crontab) while ignoring the item of interest (a Job). You did code a goodly amount of care into handling exceptions. Is there a higher level fall-back (that is, "first, do no harm" as you really don't want to stomp existing entries)? Does the crontab(1) command preserve the prior state on errors or is that something a top-level except ought do?

It is not clear what happens with non-MICRON tagged jobs, you should expect that they will exist.

One nitpick. CRONTAB_PATH is ambiguous where CRONTAB_CMD would not be. You could also put the definition of the command path at module scope which would

1. Push it higher in the source code
2. Save an unnecessary self. as module namespace is assumed and you don't expect the command to vary per instance

Finally, thanks for not making an artificial singleton of the Micron class. Not that you might have, just that some would.