build.models.MaintenanceWindow.__init__() - 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-21 22:05 UTC

build.models.MaintenanceWindow.init() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	24
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	3.0026

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	24
ccs	14
cts	15
cp	0.9333
rs	9.65
c	0
b	0
f	0
cc	3
nop	5
crap	3.0026

"""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.
        """
        self.scheduler.remove_all_jobs()
        self.scheduler.shutdown()
        windows = self.db.get_windows()

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

    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		self.scheduler.remove_all_jobs()
183	1	self.scheduler.shutdown()
184	1	windows = self.db.get_windows()
185	1
186		# Depopulate the scheduler
187	1	for window in windows:
188	1	self._unschedule(window)
189	1
190		def start_maintenance(self, mw_id: MaintenanceID):
191	1	"""Begins executing the maintenance window
192		"""
193	1	# Get Maintenance from DB and Update
194		window = self.db.start_window(mw_id)
195	1
196	1	# Activate Running
197		window.start_mw(self.controller)
198	1
199		# Schedule next task
200	1	self._schedule(window)
201	1
202		def end_maintenance(self, mw_id: MaintenanceID):
203		"""Ends execution of the maintenance window
204	1	"""
205		# Get Maintenance from DB
206		window = self.db.end_window(mw_id)
207	1
208		# Set to Ending
209	1	window.end_mw(self.controller)
210	1
211		def end_maintenance_early(self, mw_id: MaintenanceID):
212	1	"""Ends execution of the maintenance window early
213		"""
214	1	# Get Maintenance from DB
215		window = self.db.end_window(mw_id)
216
217	1	# Unschedule tasks
218		self._unschedule(window)
219
220		def add(self, window: MaintenanceWindow):
221	1	"""Add jobs to start and end a maintenance window."""
222
223	1	# Add window to DB
224	1	self.db.insert_window(window)
225	1
226	1	# Schedule next task
227	1	self._schedule(window)
228	1
229	1	def update(self, window: MaintenanceWindow):
230	1	"""Update an existing Maintenance Window."""
231
232		# Update window
233		self.db.update_window(window)
234
235		# Reschedule any pending tasks
236		self._reschedule(window)
237
238		def remove(self, mw_id: MaintenanceID):
239		"""Remove jobs that start and end a maintenance window."""
240		# Get Maintenance from DB
241		window = self.db.get_window(mw_id)
242
243		# Remove from schedule
244		self._unschedule(window)
245
246		# Remove from DB
247		self.db.remove_window(mw_id)
248
249		def _schedule(self, window: MaintenanceWindow):
250		log.info(f'Scheduling "{window.id}"')
251		if window.status == Status.PENDING:
252		self.scheduler.add_job(
253		MaintenanceStart(self, window.id),
254		'date',
255		id=f'{window.id}-start',
256		run_date=window.start
257		)
258		log.info(f'Scheduled "{window.id}" start at {window.start}')
259		if window.status == Status.RUNNING:
260		window.start_mw(self.controller)
261		self.scheduler.add_job(
262		MaintenanceEnd(self, window.id),
263		'date',
264		id=f'{window.id}-end',
265		run_date=window.end
266		)
267		log.info(f'Scheduled "{window.id}" end at {window.end}')
268
269		def _reschedule(self, window: MaintenanceWindow):
270		log.info(f'Rescheduling "{window.id}"')
271		try:
272		self.scheduler.modify_job(
273		f'{window.id}-start',
274		trigger = DateTrigger(window.start),
275		)
276		log.info(f'Rescheduled "{window.id}" start to {window.start}')
277		except JobLookupError:
278		log.info(f'Could not reschedule "{window.id}" start, no start job')
279		try:
280		self.scheduler.modify_job(
281		f'{window.id}-end',
282		trigger = DateTrigger(window.end),
283		)
284		log.info(f'Rescheduled "{window.id}" end to {window.end}')
285		except JobLookupError:
286		log.info(f'Could not reschedule "{window.id}" end, no end job')
287
288		def _unschedule(self, window: MaintenanceWindow):
289		"""Remove maintenance events from scheduler.
290		Does not update DB, due to being
291		primarily for shutdown startup cases.
292		"""
293		started = False
294		ended = False
295		try:
296		self.scheduler.remove_job(f'{window.id}-start')
297		except JobLookupError:
298		started = True
299		log.info(f'Job to start "{window.id}" already removed.')
300		try:
301		self.scheduler.remove_job(f'{window.id}-end')
302		except JobLookupError:
303		ended = True
304		log.info(f'Job to end "{window.id}" already removed.')
305		if started and not ended:
306		window.end_mw(self.controller)
307
308		def get_maintenance(self, mw_id: MaintenanceID) -> MaintenanceWindow:
309		"""Get a single maintenance by id"""
310		return self.db.get_window(mw_id)
311
312		def list_maintenances(self) -> MaintenanceWindows:
313		"""Returns a list of all maintenances"""
314		return self.db.get_windows()
315

kytos-ng / maintenance

Pull Request — master (#64)

build.models.MaintenanceWindow.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.models.MaintenanceWindow.init() A