~yerinalexey/pcrond

bafb79080c6ba6c742dbf6e43f9953b2cbd38711 — Luca Vercelli 2 years ago 8a2c122
clean code
4 files changed, 23 insertions(+), 20 deletions(-)

M pcrond/__init__.py
M pcrond/job.py
M pcrond/sched.py
M tox.ini
M pcrond/__init__.py => pcrond/__init__.py +1 -1
@@ 1,7 1,7 @@

# Here, flake8 gives error F401 '.job.Job' imported but unused
# However I have to import that, don't I ?
# pylint: disable-msg= F401 
# pylint: disable-msg=F401
from .job import Job
from .sched import Scheduler


M pcrond/job.py => pcrond/job.py +10 -10
@@ 41,10 41,10 @@ class Job(object):
            crontab = ALIASES[crontab]

        crontab_lst = crontab.split()
        

        if crontab_lst[0] in ALIASES.keys():
            raise ValueError("Cannot mix @Aliases and other tokens")
        

        if len(crontab_lst) == 5:
            crontab_lst.append("*")
        if len(crontab_lst) != 6:


@@ 120,18 120,18 @@ class Job(object):
            # here ["1","2-5","jul","10-L"]
            ranges = [x.split("-") for x in ranges]
            # here [["1"],["2","5"],["aug"], ["10","L"]]
            ranges = [[self._parse_token(w, offsets) for w in x] \
            ranges = [[self._parse_token(w, offsets) for w in x]
                                                            for x in ranges]
            # here [[1],[2,5],[7], [10,11]]
            if max([len(x) for x in ranges]) > 2:
                raise ValueError(
                    "Wrong format '%s' - a string x-y-z is meaningless" % s)
            ranges_xp = [x for x in ranges if len(x) == 1]
            ranges_xp.extend([range(x[0], x[1]+1) for x in ranges \
            ranges_xp.extend([range(x[0], x[1]+1) for x in ranges
                                             if len(x) == 2 and x[0] <= x[1]])
            ranges_xp.extend([range(x[0], maxval) for x in ranges \
            ranges_xp.extend([range(x[0], maxval) for x in ranges
                                             if len(x) == 2 and x[0] > x[1]])
            ranges_xp.extend([range(0, x[1]+1) for x in ranges \
            ranges_xp.extend([range(0, x[1]+1) for x in ranges
                                             if len(x) == 2 and x[0] > x[1]])
            # here [[2,3,4,5], [10, 11]]
            flatlist = [z for rng in ranges_xp for z in rng]


@@ 163,7 163,7 @@ class Job(object):
            import calendar
            last_day_of_month = calendar.monthrange(now.year, now.month)[1]
            if now.day == last_day_of_month:
                return True;
                return True
        return now.day in self.allowed_dom

    def __lt__(self, other):


@@ 181,12 181,12 @@ class Job(object):
        now = datetime.datetime.now()
        return not self.running \
            and (self.allowed_every_year or now.year in self.allowed_years) \
            and (self.allowed_every_month or now.month in \
            and (self.allowed_every_month or now.month in
                                                       self.allowed_months) \
            and (self.allowed_every_dow or now.weekday() in \
            and (self.allowed_every_dow or now.weekday() in
                                                 self.allowed_days_in_week) \
            and (self.allowed_every_hour or now.hour in self.allowed_hours) \
            and (self.allowed_every_minute or now.minute in \
            and (self.allowed_every_minute or now.minute in
                                                      self.allowed_minutes) \
            and (self.allowed_every_dom or self._check_day_in_month(now))


M pcrond/sched.py => pcrond/sched.py +8 -9
@@ 6,6 6,7 @@ import time

logger = logging.getLogger('schedule')


def std_launch_func(cmd_splitted):
    """
    Default way of executing commands is to invoke subprocess.run()


@@ 100,8 101,7 @@ class Scheduler(object):
        self.jobs.append(job)
        return job

    def _load_crontab_line(self, rownum, crontab_line, job_func_func=
                                                        std_launch_func):
    def _load_crontab_line(self, rownum, crontab_line, job_func_func=std_launch_func):
        """
        create a Job from a single crontab entry, and add it to this Scheduler
        :param crontab_line:


@@ 109,10 109,10 @@ class Scheduler(object):
            PRE: not empty and it not a comment
        :param job_func_func:
            function to be executed, @see load_crontab_file
        :return: a Job 
        :return: a Job
        """
        pieces = crontab_line.split()
        

        # is pattern using aliases?
        if pieces[0] in ALIASES.keys():
            try:


@@ 143,8 143,7 @@ class Scheduler(object):
                print("Error at line %d, cannot parse pattern" % rownum)
                return None

    def load_crontab_file(self, crontab_file, clear=True, job_func_func=
                                                             std_launch_func):
    def load_crontab_file(self, crontab_file, clear=True, job_func_func=std_launch_func):
        """
        Read crontab file, create corresponding jobs in this scheduler
        :param crontab_file:


@@ 157,12 156,12 @@ class Scheduler(object):
        """
        if clear:
            self.clear()
        with open(crontab_file) as fp:  
        with open(crontab_file) as fp:
            for rownum, line in enumerate(fp):
                if line is not None:	# not sure if this can happen
                if line is not None:                  # not sure if this can happen
                    line = line.strip()
                    if line != "" and line[0] != "#":
                        #skip empty lines and comments
                        # skip empty lines and comments
                        self._load_crontab_line(rownum, line, job_func_func)

    def main_loop(self):

M tox.ini => tox.ini +4 -0
@@ 6,6 6,10 @@ envlist = py27, py36
2.7.11 = py27
3.6 = py36

[pytest]
#default is 79 :(
flake8-max-line-length = 120

[testenv]
deps = -rrequirements-dev.txt
commands =