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) -> str: 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) -> None: self.deprecated = datetime.utcnow() self.deprecation_reason = reason self.updated = datetime.utcnow() def destroy(self) -> None: if self.deprecated is None: self.deprecated = datetime.utcnow() 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() ]