Passed
Pull Request — master (#140)
by Rogerio
03:53
created

build.scheduler   A

Complexity

Total Complexity 25

Size/Duplication

Total Lines 143
Duplicated Lines 0 %

Test Coverage

Coverage 92.42%

Importance

Changes 0
Metric Value
eloc 72
dl 0
loc 143
ccs 61
cts 66
cp 0.9242
rs 10
c 0
b 0
f 0
wmc 25

11 Methods

Rating   Name   Duplication   Size   Complexity  
A Scheduler.cancel_job() 0 7 2
A CircuitSchedule.id() 0 4 1
A CircuitSchedule.as_dict() 0 12 4
A CircuitSchedule.__init__() 0 9 1
A CircuitSchedule.from_dict() 0 4 1
A Scheduler.add() 0 9 2
A Scheduler.add_job() 0 28 4
A Scheduler.shutdown() 0 3 1
A Scheduler.add_circuit_job() 0 25 5
A Scheduler.__init__() 0 4 1
A Scheduler.remove() 0 4 2
1
"""Module responsible to handle schedulers."""
2 2
from uuid import uuid4
3
4 2
from apscheduler.jobstores.base import JobLookupError
5 2
from apscheduler.schedulers.background import BackgroundScheduler
6 2
from apscheduler.triggers.cron import CronTrigger
7 2
from pytz import utc
8
9 2
from kytos.core import log
10
11
12 2
class CircuitSchedule:
13
    """Schedule events."""
14
15 2
    def __init__(self, **kwargs):
16
        """Create a CircuitSchedule object."""
17 2
        self._id = kwargs.get('id', uuid4().hex)
18 2
        self.date = kwargs.get('date', None)
19
        # The minimum number of seconds to wait between retries
20 2
        self.interval = kwargs.get('interval', None)
21
        # Frequency uses Cron format. Ex: "* * * * * *"
22 2
        self.frequency = kwargs.get('frequency', None)
23 2
        self.action = kwargs.get('action', 'create')
24
25 2
    @property
26
    def id(self):  # pylint: disable=invalid-name
27
        """Return this EVC's ID."""
28 2
        return self._id
29
30 2
    @id.setter
31
    def id(self, value):  # pylint: disable=invalid-name
32
        """Set this EVC's ID."""
33 2
        self._id = value
34
35 2
    def as_dict(self):
36
        """Return a dictionary representing an circuit schedule object."""
37 2
        circuit_scheduler_dict = {'id': self.id, 'action': self.action}
38
39 2
        if self.date:
40
            circuit_scheduler_dict['date'] = self.date
41 2
        if self.frequency:
42 2
            circuit_scheduler_dict['frequency'] = self.frequency
43 2
        if self.interval:
44 2
            circuit_scheduler_dict['interval'] = self.interval
45
46 2
        return circuit_scheduler_dict
47
48 2
    @classmethod
49
    def from_dict(cls, data):
50
        """Return a CircuitSchedule object from dict."""
51 2
        return cls(**data)
52
53
54 2
class Scheduler:
55
    """Class to schedule the circuits rules."""
56
57 2
    def __init__(self):
58
        """Create a new schedule structure."""
59 2
        self.scheduler = BackgroundScheduler(timezone=utc)
60 2
        self.scheduler.start()
61
62 2
    def shutdown(self):
63
        """Shutdown the scheduler."""
64 2
        self.scheduler.shutdown(wait=False)
65
66 2
    def add(self, circuit):
67
        """
68
        Add all circuit_scheduler from specific circuit.
69
70
        Args:
71
            circuit (napps.kytos.mef_eline.models.EVCBase): EVC circuit
72
        """
73 2
        for circuit_scheduler in circuit.circuit_scheduler:
74 2
            self.add_circuit_job(circuit, circuit_scheduler)
75
76 2
    def remove(self, circuit):
77
        """Remove all scheduler from a circuit."""
78
        for job in circuit.circuit_scheduler:
79
            self.cancel_job(job.id)
80
81 2
    def add_circuit_job(self, circuit, circuit_scheduler):
82
        """
83
        Prepare the Circuit data to be added to the Scheduler.
84
85
        :param circuit(napps.kytos.mef_eline.models.EVCBase): EVC circuit
86
        :param circuit_scheduler (CircuitSchedule): Circuit schedule data
87
        :return:
88
        """
89 2
        job_call = None
90 2
        if circuit_scheduler.action == 'create':
91 2
            job_call = circuit.deploy
92 2
        elif circuit_scheduler.action == 'remove':
93 2
            job_call = circuit.remove
94
95 2
        data = {'id': circuit_scheduler.id}
96 2
        if circuit_scheduler.date:
97 2
            data.update({'run_date': circuit_scheduler.date})
98
        else:
99 2
            data.update({'start_date': circuit.start_date,
100
                         'end_date': circuit.end_date})
101
102 2
        if circuit_scheduler.interval:
103 2
            data.update(circuit_scheduler.interval)
104
105 2
        self.add_job(circuit_scheduler, job_call, data)
106
107 2
    def add_job(self, circuit_scheduler, job_call, data):
108
        """
109
        Add a specific cron job to the scheduler.
110
111
        Args:
112
            circuit_scheduler: CircuitSchedule object
113
            job_call: function to be called by the job
114
            data: Dict to pass to the job_call as parameter
115
                if job_call is a date, the template is like:
116
                 {'id': <ID>, 'run_date': date } or
117
                 {'id': <ID>, 'start_date': date, 'end_date': date }
118
                if job_call is an interval, the template is like:
119
                    {   'id': <ID>,
120
                        'hours': 2,
121
                        'minutes': 3
122
                    }
123
                if job_call is frequency, the template is the cron format.
124
        """
125 2
        if circuit_scheduler.date:
126 2
            self.scheduler.add_job(job_call, 'date', **data)
127
128 2
        elif circuit_scheduler.interval:
129 2
            self.scheduler.add_job(job_call, 'interval', **data)
130
131 2
        elif circuit_scheduler.frequency:
132 2
            cron = CronTrigger.from_crontab(circuit_scheduler.frequency,
133
                                            timezone=utc)
134 2
            self.scheduler.add_job(job_call, cron, **data)
135
136 2
    def cancel_job(self, circuit_scheduler_id):
137
        """Cancel a specific job from scheduler."""
138 2
        try:
139 2
            self.scheduler.remove_job(circuit_scheduler_id)
140
        except JobLookupError as job_error:
141
            # Job was not found... Maybe someone already remove it.
142
            log.error("Scheduler error cancelling job. %s" % job_error)
143