Commits

Anonymous committed 37ef965

Initial import

  • Participants

Comments (0)

Files changed (2)

+# use glob syntax.
+syntax: glob
+*.pyc
+*.orig
+*.swp
+#-*- 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()