119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
import logging
|
|
from abc import abstractmethod
|
|
from datetime import datetime
|
|
from typing import Union, List, Optional, Any
|
|
|
|
from app.alarms import alarms_for
|
|
from app.brm.brn import BRN
|
|
from app.extensions import db
|
|
from app.models.alarms import Alarm
|
|
|
|
|
|
class AbstractConfiguration(db.Model): # type: ignore
|
|
__abstract__ = True
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
description = db.Column(db.String(255), nullable=False)
|
|
added = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
|
|
updated = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
|
|
destroyed = db.Column(db.DateTime(), nullable=True)
|
|
|
|
@property
|
|
def alarms(self) -> List[Alarm]:
|
|
return alarms_for(self.brn)
|
|
|
|
@property
|
|
@abstractmethod
|
|
def brn(self) -> BRN:
|
|
raise NotImplementedError()
|
|
|
|
def destroy(self) -> None:
|
|
self.destroyed = datetime.utcnow()
|
|
self.updated = datetime.utcnow()
|
|
|
|
@classmethod
|
|
def csv_header(cls) -> List[str]:
|
|
return [
|
|
"id", "description", "added", "updated", "destroyed"
|
|
]
|
|
|
|
def csv_row(self) -> List[Any]:
|
|
return [
|
|
getattr(self, x) for x in self.csv_header()
|
|
]
|
|
|
|
|
|
class AbstractResource(db.Model): # type: ignore
|
|
__abstract__ = True
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
added = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
|
|
updated = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
|
|
deprecated = db.Column(db.DateTime(), nullable=True)
|
|
deprecation_reason = db.Column(db.String(), nullable=True)
|
|
destroyed = db.Column(db.DateTime(), nullable=True)
|
|
|
|
def __init__(self, *,
|
|
id: Optional[int] = None,
|
|
added: Optional[datetime] = None,
|
|
updated: Optional[datetime] = None,
|
|
deprecated: Optional[datetime] = None,
|
|
deprecation_reason: Optional[str] = None,
|
|
destroyed: Optional[datetime] = None,
|
|
**kwargs: Any) -> None:
|
|
super().__init__(id=id,
|
|
added=added,
|
|
updated=updated,
|
|
deprecated=deprecated,
|
|
deprecation_reason=deprecation_reason,
|
|
destroyed=destroyed,
|
|
**kwargs)
|
|
if self.added is None:
|
|
self.added = datetime.utcnow()
|
|
if self.updated is None:
|
|
self.updated = datetime.utcnow()
|
|
|
|
@property
|
|
@abstractmethod
|
|
def brn(self) -> BRN:
|
|
raise NotImplementedError()
|
|
|
|
def deprecate(self, *, reason: str) -> bool:
|
|
"""
|
|
Marks the resource as deprecated. In the event that the resource was already
|
|
deprecated, no change will be recorded and the function will return False.
|
|
|
|
:param reason: an opaque string that records the deprecation reason
|
|
:return: if the proxy was deprecated
|
|
"""
|
|
if self.deprecated is None:
|
|
logging.info("Deprecating %s (reason=%s)", self.brn, reason)
|
|
self.deprecated = datetime.utcnow()
|
|
self.deprecation_reason = reason
|
|
self.updated = datetime.utcnow()
|
|
return True
|
|
else:
|
|
logging.info("Not deprecating %s (reason=%s) because it's already deprecated", self.brn, reason)
|
|
return False
|
|
|
|
def destroy(self) -> None:
|
|
"""
|
|
Marks the resource for destruction.
|
|
|
|
:return: None
|
|
"""
|
|
if self.deprecated is None:
|
|
self.deprecate(reason="destroyed")
|
|
self.destroyed = datetime.utcnow()
|
|
self.updated = datetime.utcnow()
|
|
|
|
@classmethod
|
|
def csv_header(cls) -> List[str]:
|
|
return [
|
|
"id", "added", "updated", "deprecated", "deprecation_reason", "destroyed"
|
|
]
|
|
|
|
def csv_row(self) -> List[Union[datetime, bool, int, str]]:
|
|
return [
|
|
getattr(self, x) for x in self.csv_header()
|
|
]
|