build.models.MaintenanceWindow.str_to_datetime() - 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 2023-01-03 01:46 UTC

build.models.MaintenanceWindow.str_to_datetime() A

↳ Parent: build.models

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	5
ccs	4
cts	4
cp	1
rs	10
c	0
b	0
f	0
cc	1
nop	1
crap	1

"""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: key not in values or 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]

    def __len__(self):
        return len(self.__root__)

    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)

class OverlapError(Exception):
    """
    Exception for when a Maintenance Windows execution
    period overlaps with one or more windows.
    """
    new_window: MaintenanceWindow
    interferring: MaintenanceWindows

    def __init__(self, new_window: MaintenanceWindow, interferring: MaintenanceWindows):
        self.new_window = new_window
        self.interferring = interferring

    def __str__(self):
        return f"Maintenance Window '{self.new_window.id}'<{self.new_window.start} to {self.new_window.end}> " +\
            "interfers with the following windows: " +\
            '[' +\
            ', '.join([f"'{window.id}'<{window.start} to {window.end}>" for window in self.interferring]) +\
            ']'

@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, force = False):
        """Add jobs to start and end a maintenance window."""

        if force is False:
            overlapping_windows = self.db.check_overlap(window)
            if overlapping_windows:
                raise OverlapError(window, overlapping_windows)

        # 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: key not in values or 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	def __len__(self):
119	1	return len(self.__root__)
120	1
121	1	class Config:
122		json_encoders = {
123	1	datetime: lambda v: v.strftime(TIME_FMT),
124		}
125	1
126	1	@dataclass
127	1	class MaintenanceStart:
128	1	"""
129	1	Callable used for starting maintenance windows
130	1	"""
131	1	maintenance_scheduler: 'Scheduler'
132		mw_id: MaintenanceID
133
134	1	def __call__(self):
135	1	self.maintenance_scheduler.start_maintenance(self.mw_id)
136
137
138		@dataclass
139		class MaintenanceEnd:
140	1	"""
141	1	Callable used for ending maintenance windows
142		"""
143		maintenance_scheduler: 'Scheduler'
144		mw_id: MaintenanceID
145
146		def __call__(self):
147		self.maintenance_scheduler.end_maintenance(self.mw_id)
148
149		class OverlapError(Exception):
150	1	"""
151	1	Exception for when a Maintenance Windows execution
152		period overlaps with one or more windows.
153		"""
154		new_window: MaintenanceWindow
155		interferring: MaintenanceWindows
156
157		def __init__(self, new_window: MaintenanceWindow, interferring: MaintenanceWindows):
158		self.new_window = new_window
159		self.interferring = interferring
160
161		def __str__(self):
162		return f"Maintenance Window '{self.new_window.id}'<{self.new_window.start} to {self.new_window.end}> " +\
163		"interfers with the following windows: " +\
164		'[' +\
165		', '.join([f"'{window.id}'<{window.start} to {window.end}>" for window in self.interferring]) +\
166		']'
167	1
168	1	@dataclass
169		class Scheduler:
170	1	"""Scheduler for a maintenance window."""
171	1	controller: Controller
172		db: 'MaintenanceController'
173	1	scheduler: BaseScheduler
174
175	1	@classmethod
176	1	def new_scheduler(cls, controller: Controller):
177	1	"""
178	1	Creates a new scheduler from the given kytos controller
179	1	"""
180	1	scheduler = BackgroundScheduler(timezone=pytz.utc)
181	1	from napps.kytos.maintenance.controllers import MaintenanceController
182		db = MaintenanceController()
183	1	db.bootstrap_indexes()
184	1	instance = cls(controller, db, scheduler)
185	1	return instance
186
187	1	def start(self):
188	1	"""
189	1	Begin running the scheduler.
190		"""
191	1	# Populate the scheduler with all pending tasks
192		windows = self.db.get_windows()
193	1	for window in windows:
194		self._schedule(window)
195	1
196	1	# Start the scheduler
197		self.scheduler.start()
198	1
199		def shutdown(self):
200	1	"""
201	1	Stop running the scheduler.
202		"""
203		windows = self.db.get_windows()
204	1
205		# Depopulate the scheduler
206		for window in windows:
207	1	self._unschedule(window)
208
209	1	self.scheduler.remove_all_jobs()
210	1	self.scheduler.shutdown()
211
212	1	def start_maintenance(self, mw_id: MaintenanceID):
213		"""Begins executing the maintenance window
214	1	"""
215		# Get Maintenance from DB and Update
216		window = self.db.start_window(mw_id)
217	1
218		# Activate Running
219		window.start_mw(self.controller)
220
221	1	# Schedule next task
222		self._schedule(window)
223	1
224	1	def end_maintenance(self, mw_id: MaintenanceID):
225	1	"""Ends execution of the maintenance window
226	1	"""
227	1	# Get Maintenance from DB
228	1	window = self.db.end_window(mw_id)
229	1
230	1	# Set to Ending
231		window.end_mw(self.controller)
232
233		def end_maintenance_early(self, mw_id: MaintenanceID):
234		"""Ends execution of the maintenance window early
235		"""
236		# Get Maintenance from DB
237		window = self.db.end_window(mw_id)
238
239		# Unschedule tasks
240		self._unschedule(window)
241
242		def add(self, window: MaintenanceWindow, force = False):
243		"""Add jobs to start and end a maintenance window."""
244
245		if force is False:
246		overlapping_windows = self.db.check_overlap(window)
247		if overlapping_windows:
248		raise OverlapError(window, overlapping_windows)
249
250		# Add window to DB
251		self.db.insert_window(window)
252
253		# Schedule next task
254		self._schedule(window)
255
256		def update(self, window: MaintenanceWindow):
257		"""Update an existing Maintenance Window."""
258
259		# Update window
260		self.db.update_window(window)
261
262		# Reschedule any pending tasks
263		self._reschedule(window)
264
265		def remove(self, mw_id: MaintenanceID):
266		"""Remove jobs that start and end a maintenance window."""
267		# Get Maintenance from DB
268		window = self.db.get_window(mw_id)
269
270		# Remove from schedule
271		self._unschedule(window)
272
273		# Remove from DB
274		self.db.remove_window(mw_id)
275
276		def _schedule(self, window: MaintenanceWindow):
277		log.info(f'Scheduling "{window.id}"')
278		if window.status == Status.PENDING:
279		self.scheduler.add_job(
280		MaintenanceStart(self, window.id),
281		'date',
282		id=f'{window.id}-start',
283		run_date=window.start
284		)
285		log.info(f'Scheduled "{window.id}" start at {window.start}')
286		if window.status == Status.RUNNING:
287		window.start_mw(self.controller)
288		self.scheduler.add_job(
289		MaintenanceEnd(self, window.id),
290		'date',
291		id=f'{window.id}-end',
292		run_date=window.end
293		)
294		log.info(f'Scheduled "{window.id}" end at {window.end}')
295
296		def _reschedule(self, window: MaintenanceWindow):
297		log.info(f'Rescheduling "{window.id}"')
298		try:
299		self.scheduler.modify_job(
300		f'{window.id}-start',
301		trigger = DateTrigger(window.start),
302		)
303		log.info(f'Rescheduled "{window.id}" start to {window.start}')
304		except JobLookupError:
305		log.info(f'Could not reschedule "{window.id}" start, no start job')
306		try:
307		self.scheduler.modify_job(
308		f'{window.id}-end',
309		trigger = DateTrigger(window.end),
310		)
311		log.info(f'Rescheduled "{window.id}" end to {window.end}')
312		except JobLookupError:
313		log.info(f'Could not reschedule "{window.id}" end, no end job')
314
315		def _unschedule(self, window: MaintenanceWindow):
316		"""Remove maintenance events from scheduler.
317		Does not update DB, due to being
318		primarily for shutdown startup cases.
319		"""
320		started = False
321		ended = False
322		try:
323		self.scheduler.remove_job(f'{window.id}-start')
324		except JobLookupError:
325		started = True
326		log.info(f'Job to start "{window.id}" already removed.')
327		try:
328		self.scheduler.remove_job(f'{window.id}-end')
329		except JobLookupError:
330		ended = True
331		log.info(f'Job to end "{window.id}" already removed.')
332		if started and not ended:
333		window.end_mw(self.controller)
334
335		def get_maintenance(self, mw_id: MaintenanceID) -> MaintenanceWindow:
336		"""Get a single maintenance by id"""
337		return self.db.get_window(mw_id)
338
339		def list_maintenances(self) -> MaintenanceWindows:
340		"""Returns a list of all maintenances"""
341		return self.db.get_windows()
342

kytos-ng / maintenance

Pull Request — master (#64)

build.models.MaintenanceWindow.str_to_datetime() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like