import logging from abc import abstractmethod from datetime import datetime from typing import Union, List, Optional, Any from app.brm.brn import BRN from app.extensions import db 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 @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() ]