Комментарий запрос: простая бандероль хрон в Python


Это моя первая попытка создать многоразовые и протестирован модуль. Любые комментарии высоко ценится.

#-*- 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()


623
3
задан 7 июня 2011 в 11:06 Источник Поделиться
Комментарии
2 ответа

Я думаю, что это довольно хорошее начало:


  • Я хотел бы видеть работу класса.

  • Выходной поток stderr будет большое в случаях retcode != 0.

  • Проверка и/или выхода из программы и все было бы отлично.

2
ответ дан 7 июня 2011 в 04:06 Источник Поделиться

Я согласен с dmeister.

Задания обрабатываются слишком непрозрачно на Микрон и, следовательно, не имеют доступа. Crontab так высоко структурирована и хорошую работу позволит проверка и модификация ее элементов.

Это заставляет меня думать, что вы сосредоточены на контейнер (кронтаб), игнорируя пункт интереса (работа). Вы сделали код солидная сумма помощи в обработке исключений. Есть высшее откат уровне (то есть, "не навреди", так как вы действительно не хотите топать существующие записи)? Тут кронтаб(1) команду сохранить предыдущее состояние на ошибки или что-то топ-уровня, кроме следует сделать?

Непонятно, что происходит с non-микрон помечены вакансии, вы должны ожидать, что они будут существовать.

Одним придираться. CRONTAB_PATH неоднозначно, где CRONTAB_CMD не будет. Кроме того, можно поставить определение командной путь в области модуля, который бы


  1. Подтолкнуть его выше в исходном коде

  2. Сохранить ненужные самостоятельно. поскольку пространство имен модуля предполагается и не ожидать, что команда варьироваться в зависимости от экземпляра

Наконец-то, спасибо, не делая искусственное одноэлементного класса Микрона. Не то, что у вас есть, только то, что некоторые.

1
ответ дан 17 июня 2011 в 04:06 Источник Поделиться