build.models.MaintenanceWindow.update() - Code Metrics - Inspection of "Add in persistence to maintenance windows" - kytos-ng/maintenance - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#64)

unknown

created 2022-12-27 15:53 UTC

build.models.MaintenanceWindow.update() B

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	23
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	19
CRAP Score	8.1624

Importance

Changes

Metric	Value
eloc	22
dl	0
loc	23
ccs	19
cts	22
cp	0.8636
rs	7.3333
c	0
b	0
f	0
cc	8
nop	2
crap	8.1624

"""Models used by the maintenance NApp.

This module define models for the maintenance window itself and the
scheduler.
"""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import NewType, Optional
from uuid import uuid4

import pytz
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.base import BaseScheduler
from apscheduler.triggers.date import DateTrigger
from pydantic import BaseModel, Field, validator, root_validator

from kytos.core import KytosEvent, log
from kytos.core.controller import Controller

TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"

class Status(str, Enum):
    """Maintenance windows status."""

    PENDING = 'pending'
    RUNNING = 'running'
    FINISHED = 'finished'


MaintenanceID = NewType('MaintenanceID', str)


class MaintenanceWindow(BaseModel):
    """Class for structure of maintenance windows.
    """
    start: datetime
    end: datetime
    switches: list[str] = Field(default_factory = list)
    interfaces: list[str] = Field(default_factory = list)
    links: list[str] = Field(default_factory = list)
    id: MaintenanceID = Field(
        default_factory = lambda: MaintenanceID(uuid4().hex)
    )
    description: str = Field(default = '')
    status: Status = Field(default=Status.PENDING)
    inserted_at: Optional[datetime] = Field(default = None)
    updated_at: Optional[datetime] = Field(default = None)

    @validator('start', 'end', pre = True)
    def convert_time(cls, time):
        if isinstance(time, str):
            time = datetime.strptime(time, TIME_FMT)
        return time

    @validator('start')
    def check_start_in_past(cls, start_time):
        if start_time < datetime.now(pytz.utc):
            raise ValueError('Start in the past not allowed')
        return start_time

    @validator('end')
    def check_end_before_start(cls, end_time, values):
        if 'start' in values and end_time <= values['start']:
            raise ValueError('End before start not allowed')
        return end_time
    
    @root_validator
    def check_items_empty(cls, values):
        if all(map(lambda key: len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
            raise ValueError('At least one item must be provided')
        return values

    def maintenance_event(self, operation, controller: Controller):
        """Create events to start/end a maintenance."""
        if self.switches:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_switch',
                content={'switches': self.switches}
            )
            controller.buffers.app.put(event)
        if self.interfaces:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_interface',
                content={'unis': self.interfaces}
            )
            controller.buffers.app.put(event)
        if self.links:
            event = KytosEvent(
                name=f'kytos/maintenance.{operation}_link',
                content={'links': self.links}
            )
            controller.buffers.app.put(event)

    def start_mw(self, controller: Controller):
        """Actions taken when a maintenance window starts."""
        self.maintenance_event('start', controller)

    def end_mw(self, controller: Controller):
        """Actions taken when a maintenance window finishes."""
        self.maintenance_event('end', controller)

    class Config:
        json_encoders = {
            datetime: lambda v: v.strftime(TIME_FMT),
        }

class MaintenanceWindows(BaseModel):
    __root__: list[MaintenanceWindow]

    def __iter__(self):
        return iter(self.__root__)

    def __getitem__(self, item):
        return self.__root__[item]

    class Config:
        json_encoders = {
            datetime: lambda v: v.strftime(TIME_FMT),
        }

@dataclass
class MaintenanceStart:
    """
    Callable used for starting maintenance windows
    """
    maintenance_scheduler: 'Scheduler'
    mw_id: MaintenanceID

    def __call__(self):
        self.maintenance_scheduler.start_maintenance(self.mw_id)


@dataclass
class MaintenanceEnd:
    """
    Callable used for ending maintenance windows
    """
    maintenance_scheduler: 'Scheduler'
    mw_id: MaintenanceID

    def __call__(self):
        self.maintenance_scheduler.end_maintenance(self.mw_id)


@dataclass
class Scheduler:
    """Scheduler for a maintenance window."""
    controller: Controller
    db: 'MaintenanceController'
    scheduler: BaseScheduler

    @classmethod
    def new_scheduler(cls, controller: Controller):
        """
        Creates a new scheduler from the given kytos controller
        """
        scheduler = BackgroundScheduler(timezone=pytz.utc)
        from napps.kytos.maintenance.controllers import MaintenanceController
        db = MaintenanceController()
        db.bootstrap_indexes()
        instance = cls(controller, db, scheduler)
        return instance

    def start(self):
        """
        Begin running the scheduler.
        """
        # Populate the scheduler with all pending tasks
        windows = self.db.get_windows()
        for window in windows:
            self._schedule(window)

        # Start the scheduler
        self.scheduler.start()

    def shutdown(self):
        """
        Stop running the scheduler.
        """
        windows = self.db.get_windows()

        # Depopulate the scheduler
        for window in windows:
            self._unschedule(window)

        self.scheduler.remove_all_jobs()
        self.scheduler.shutdown()

    def start_maintenance(self, mw_id: MaintenanceID):
        """Begins executing the maintenance window
        """
        # Get Maintenance from DB and Update
        window = self.db.start_window(mw_id)

        # Activate Running
        window.start_mw(self.controller)

        # Schedule next task
        self._schedule(window)

    def end_maintenance(self, mw_id: MaintenanceID):
        """Ends execution of the maintenance window
        """
        # Get Maintenance from DB
        window = self.db.end_window(mw_id)

        # Set to Ending
        window.end_mw(self.controller)

    def end_maintenance_early(self, mw_id: MaintenanceID):
        """Ends execution of the maintenance window early
        """
        # Get Maintenance from DB
        window = self.db.end_window(mw_id)

        # Unschedule tasks
        self._unschedule(window)

    def add(self, window: MaintenanceWindow):
        """Add jobs to start and end a maintenance window."""

        # Add window to DB
        self.db.insert_window(window)

        # Schedule next task
        self._schedule(window)

    def update(self, window: MaintenanceWindow):
        """Update an existing Maintenance Window."""

        # Update window
        self.db.update_window(window)

        # Reschedule any pending tasks
        self._reschedule(window)

    def remove(self, mw_id: MaintenanceID):
        """Remove jobs that start and end a maintenance window."""
        # Get Maintenance from DB
        window = self.db.get_window(mw_id)

        # Remove from schedule
        self._unschedule(window)

        # Remove from DB
        self.db.remove_window(mw_id)

    def _schedule(self, window: MaintenanceWindow):
        log.info(f'Scheduling "{window.id}"')
        if window.status == Status.PENDING:
            self.scheduler.add_job(
                MaintenanceStart(self, window.id),
                'date',
                id=f'{window.id}-start',
                run_date=window.start
            )
            log.info(f'Scheduled "{window.id}" start at {window.start}')
        if window.status == Status.RUNNING:
            window.start_mw(self.controller)
            self.scheduler.add_job(
                MaintenanceEnd(self, window.id),
                'date',
                id=f'{window.id}-end',
                run_date=window.end
            )
            log.info(f'Scheduled "{window.id}" end at {window.end}')

    def _reschedule(self, window: MaintenanceWindow):
        log.info(f'Rescheduling "{window.id}"')
        try:
            self.scheduler.modify_job(
                f'{window.id}-start',
                trigger = DateTrigger(window.start),
            )
            log.info(f'Rescheduled "{window.id}" start to {window.start}')
        except JobLookupError:
            log.info(f'Could not reschedule "{window.id}" start, no start job')
        try:
            self.scheduler.modify_job(
                f'{window.id}-end',
                trigger = DateTrigger(window.end),
            )
            log.info(f'Rescheduled "{window.id}" end to {window.end}')
        except JobLookupError:
            log.info(f'Could not reschedule "{window.id}" end, no end job')

    def _unschedule(self, window: MaintenanceWindow):
        """Remove maintenance events from scheduler.
        Does not update DB, due to being
        primarily for shutdown startup cases.
        """
        started = False
        ended = False
        try:
            self.scheduler.remove_job(f'{window.id}-start')
        except JobLookupError:
            started = True
            log.info(f'Job to start "{window.id}" already removed.')
        try:
            self.scheduler.remove_job(f'{window.id}-end')
        except JobLookupError:
            ended = True
            log.info(f'Job to end "{window.id}" already removed.')
        if started and not ended:
            window.end_mw(self.controller)

    def get_maintenance(self, mw_id: MaintenanceID) -> MaintenanceWindow:
        """Get a single maintenance by id"""
        return self.db.get_window(mw_id)

    def list_maintenances(self) -> MaintenanceWindows:
        """Returns a list of all maintenances"""
        return self.db.get_windows()


1		"""Models used by the maintenance NApp.
2
3		This module define models for the maintenance window itself and the
4		scheduler.
5		"""
6	1	from dataclasses import dataclass
7	1	from datetime import datetime
8	1	from enum import Enum
9		from typing import NewType, Optional
10	1	from uuid import uuid4
11	1
12	1	import pytz
13		from apscheduler.jobstores.base import JobLookupError
14	1	from apscheduler.schedulers.background import BackgroundScheduler
15	1	from apscheduler.schedulers.base import BaseScheduler
16	1	from apscheduler.triggers.date import DateTrigger
17		from pydantic import BaseModel, Field, validator, root_validator
18	1
19		from kytos.core import KytosEvent, log
20		from kytos.core.controller import Controller
21	1
22		TIME_FMT = "%Y-%m-%dT%H:%M:%S%z"
23
24	1	class Status(str, Enum):
25	1	"""Maintenance windows status."""
26	1
27		PENDING = 'pending'
28		RUNNING = 'running'
29	1	FINISHED = 'finished'
30
31
32	1	MaintenanceID = NewType('MaintenanceID', str)
33
34
35		class MaintenanceWindow(BaseModel):
36		"""Class for structure of maintenance windows.
37		"""
38		start: datetime
39		end: datetime
40		switches: list[str] = Field(default_factory = list)
41		interfaces: list[str] = Field(default_factory = list)
42	1	links: list[str] = Field(default_factory = list)
43	1	id: MaintenanceID = Field(
44	1	default_factory = lambda: MaintenanceID(uuid4().hex)
45		)
46	1	description: str = Field(default = '')
47	1	status: Status = Field(default=Status.PENDING)
48	1	inserted_at: Optional[datetime] = Field(default = None)
49	1	updated_at: Optional[datetime] = Field(default = None)
50	1
51	1	@validator('start', 'end', pre = True)
52	1	def convert_time(cls, time):
53	1	if isinstance(time, str):
54	1	time = datetime.strptime(time, TIME_FMT)
55	1	return time
56
57	1	@validator('start')
58	1	def check_start_in_past(cls, start_time):
59		if start_time < datetime.now(pytz.utc):
60	1	raise ValueError('Start in the past not allowed')
61		return start_time
62	1
63	1	@validator('end')
64		def check_end_before_start(cls, end_time, values):
65	1	if 'start' in values and end_time <= values['start']:
66	1	raise ValueError('End before start not allowed')
67	1	return end_time
68	1
69	1	@root_validator
70		def check_items_empty(cls, values):
71	1	if all(map(lambda key: len(values[key]) == 0, ['switches', 'links', 'interfaces'])):
72		raise ValueError('At least one item must be provided')
73		return values
74	1
75		def maintenance_event(self, operation, controller: Controller):
76	1	"""Create events to start/end a maintenance."""
77		if self.switches:
78	1	event = KytosEvent(
79	1	name=f'kytos/maintenance.{operation}_switch',
80	1	content={'switches': self.switches}
81	1	)
82	1	controller.buffers.app.put(event)
83	1	if self.interfaces:
84	1	event = KytosEvent(
85	1	name=f'kytos/maintenance.{operation}_interface',
86	1	content={'unis': self.interfaces}
87	1	)
88	1	controller.buffers.app.put(event)
89	1	if self.links:
90	1	event = KytosEvent(
91		name=f'kytos/maintenance.{operation}_link',
92	1	content={'links': self.links}
93	1	)
94		controller.buffers.app.put(event)
95
96		def start_mw(self, controller: Controller):
97		"""Actions taken when a maintenance window starts."""
98		self.maintenance_event('start', controller)
99
100		def end_mw(self, controller: Controller):
101		"""Actions taken when a maintenance window finishes."""
102		self.maintenance_event('end', controller)
103
104		class Config:
105		json_encoders = {
106		datetime: lambda v: v.strftime(TIME_FMT),
107		}
108
109		class MaintenanceWindows(BaseModel):
110	1	__root__: list[MaintenanceWindow]
111
112	1	def __iter__(self):
113	1	return iter(self.__root__)
114	1
115	1	def __getitem__(self, item):
116	1	return self.__root__[item]
117	1
118	1	class Config:
119	1	json_encoders = {
120	1	datetime: lambda v: v.strftime(TIME_FMT),
121	1	}
122
123	1	@dataclass
124		class MaintenanceStart:
125	1	"""
126	1	Callable used for starting maintenance windows
127	1	"""
128	1	maintenance_scheduler: 'Scheduler'
129	1	mw_id: MaintenanceID
130	1
131	1	def __call__(self):
132		self.maintenance_scheduler.start_maintenance(self.mw_id)
133
134	1
135	1	@dataclass
136		class MaintenanceEnd:
137		"""
138		Callable used for ending maintenance windows
139		"""
140	1	maintenance_scheduler: 'Scheduler'
141	1	mw_id: MaintenanceID
142
143		def __call__(self):
144		self.maintenance_scheduler.end_maintenance(self.mw_id)
145
146
147		@dataclass
148		class Scheduler:
149		"""Scheduler for a maintenance window."""
150	1	controller: Controller
151	1	db: 'MaintenanceController'
152		scheduler: BaseScheduler
153
154		@classmethod
155		def new_scheduler(cls, controller: Controller):
156		"""
157		Creates a new scheduler from the given kytos controller
158		"""
159		scheduler = BackgroundScheduler(timezone=pytz.utc)
160		from napps.kytos.maintenance.controllers import MaintenanceController
161		db = MaintenanceController()
162		db.bootstrap_indexes()
163		instance = cls(controller, db, scheduler)
164		return instance
165
166		def start(self):
167	1	"""
168	1	Begin running the scheduler.
169		"""
170	1	# Populate the scheduler with all pending tasks
171	1	windows = self.db.get_windows()
172		for window in windows:
173	1	self._schedule(window)
174
175	1	# Start the scheduler
176	1	self.scheduler.start()
177	1
178	1	def shutdown(self):
179	1	"""
180	1	Stop running the scheduler.
181	1	"""
182		windows = self.db.get_windows()
183	1
184	1	# Depopulate the scheduler
185	1	for window in windows:
186		self._unschedule(window)
187	1
188	1	self.scheduler.remove_all_jobs()
189	1	self.scheduler.shutdown()
190
191	1	def start_maintenance(self, mw_id: MaintenanceID):
192		"""Begins executing the maintenance window
193	1	"""
194		# Get Maintenance from DB and Update
195	1	window = self.db.start_window(mw_id)
196	1
197		# Activate Running
198	1	window.start_mw(self.controller)
199
200	1	# Schedule next task
201	1	self._schedule(window)
202
203		def end_maintenance(self, mw_id: MaintenanceID):
204	1	"""Ends execution of the maintenance window
205		"""
206		# Get Maintenance from DB
207	1	window = self.db.end_window(mw_id)
208
209	1	# Set to Ending
210	1	window.end_mw(self.controller)
211
212	1	def end_maintenance_early(self, mw_id: MaintenanceID):
213		"""Ends execution of the maintenance window early
214	1	"""
215		# Get Maintenance from DB
216		window = self.db.end_window(mw_id)
217	1
218		# Unschedule tasks
219		self._unschedule(window)
220
221	1	def add(self, window: MaintenanceWindow):
222		"""Add jobs to start and end a maintenance window."""
223	1
224	1	# Add window to DB
225	1	self.db.insert_window(window)
226	1
227	1	# Schedule next task
228	1	self._schedule(window)
229	1
230	1	def update(self, window: MaintenanceWindow):
231		"""Update an existing Maintenance Window."""
232
233		# Update window
234		self.db.update_window(window)
235
236		# Reschedule any pending tasks
237		self._reschedule(window)
238
239		def remove(self, mw_id: MaintenanceID):
240		"""Remove jobs that start and end a maintenance window."""
241		# Get Maintenance from DB
242		window = self.db.get_window(mw_id)
243
244		# Remove from schedule
245		self._unschedule(window)
246
247		# Remove from DB
248		self.db.remove_window(mw_id)
249
250		def _schedule(self, window: MaintenanceWindow):
251		log.info(f'Scheduling "{window.id}"')
252		if window.status == Status.PENDING:
253		self.scheduler.add_job(
254		MaintenanceStart(self, window.id),
255		'date',
256		id=f'{window.id}-start',
257		run_date=window.start
258		)
259		log.info(f'Scheduled "{window.id}" start at {window.start}')
260		if window.status == Status.RUNNING:
261		window.start_mw(self.controller)
262		self.scheduler.add_job(
263		MaintenanceEnd(self, window.id),
264		'date',
265		id=f'{window.id}-end',
266		run_date=window.end
267		)
268		log.info(f'Scheduled "{window.id}" end at {window.end}')
269
270		def _reschedule(self, window: MaintenanceWindow):
271		log.info(f'Rescheduling "{window.id}"')
272		try:
273		self.scheduler.modify_job(
274		f'{window.id}-start',
275		trigger = DateTrigger(window.start),
276		)
277		log.info(f'Rescheduled "{window.id}" start to {window.start}')
278		except JobLookupError:
279		log.info(f'Could not reschedule "{window.id}" start, no start job')
280		try:
281		self.scheduler.modify_job(
282		f'{window.id}-end',
283		trigger = DateTrigger(window.end),
284		)
285		log.info(f'Rescheduled "{window.id}" end to {window.end}')
286		except JobLookupError:
287		log.info(f'Could not reschedule "{window.id}" end, no end job')
288
289		def _unschedule(self, window: MaintenanceWindow):
290		"""Remove maintenance events from scheduler.
291		Does not update DB, due to being
292		primarily for shutdown startup cases.
293		"""
294		started = False
295		ended = False
296		try:
297		self.scheduler.remove_job(f'{window.id}-start')
298		except JobLookupError:
299		started = True
300		log.info(f'Job to start "{window.id}" already removed.')
301		try:
302		self.scheduler.remove_job(f'{window.id}-end')
303		except JobLookupError:
304		ended = True
305		log.info(f'Job to end "{window.id}" already removed.')
306		if started and not ended:
307		window.end_mw(self.controller)
308
309		def get_maintenance(self, mw_id: MaintenanceID) -> MaintenanceWindow:
310		"""Get a single maintenance by id"""
311		return self.db.get_window(mw_id)
312
313		def list_maintenances(self) -> MaintenanceWindows:
314		"""Returns a list of all maintenances"""
315		return self.db.get_windows()
316

kytos-ng / maintenance

Pull Request — master (#64)

build.models.MaintenanceWindow.update() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like