majuna/app/models/__init__.py

import logging
from abc import abstractmethod
from datetime import datetime
from typing import Union, List, Optional, Any, Dict

from sqlalchemy.orm import Mapped, mapped_column

from app.brm.brn import BRN
from app.extensions import db


class AbstractConfiguration(db.Model):  # type: ignore
    __abstract__ = True

    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    description: Mapped[str]
    added: Mapped[datetime]
    updated: Mapped[datetime]
    destroyed: Mapped[Optional[datetime]]

    @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 Deprecation(db.Model):  # type: ignore[name-defined,misc]
    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    resource_type: Mapped[str] = mapped_column(db.String(50))
    resource_id: Mapped[int] = mapped_column(db.Integer)
    deprecated_at: Mapped[datetime] = mapped_column(db.DateTime(), default=datetime.utcnow, nullable=False)
    meta: Mapped[Optional[Dict[str, Any]]] = mapped_column(db.JSON())
    reason: Mapped[str] = mapped_column(db.String(), nullable=False)

    @property
    def resource(self) -> "AbstractResource":
        from app.models.mirrors import Proxy  # pylint: disable=R0401
        model = {'Proxy': Proxy}[self.resource_type]
        return model.query.get(self.resource_id)  # type: ignore[no-any-return]


class AbstractResource(db.Model):  # type: ignore
    __abstract__ = True

    id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
    added: Mapped[datetime] = mapped_column(db.DateTime(), default=datetime.utcnow, nullable=False)
    updated: Mapped[datetime] = mapped_column(db.DateTime(), default=datetime.utcnow, nullable=False)
    deprecated: Mapped[Optional[datetime]] = mapped_column(db.DateTime())
    deprecation_reason: Mapped[Optional[str]] = mapped_column(db.String())
    destroyed: Mapped[Optional[datetime]] = mapped_column(db.DateTime())

    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:
        if added is None:
            added = datetime.utcnow()
        if updated is None:
            updated = datetime.utcnow()
        super().__init__(id=id,
                         added=added,
                         updated=updated,
                         deprecated=deprecated,
                         deprecation_reason=deprecation_reason,
                         destroyed=destroyed,
                         **kwargs)

    @property
    @abstractmethod
    def brn(self) -> BRN:
        raise NotImplementedError()

    def deprecate(self, *, reason: str, meta: Optional[Dict[str, Any]] = None) -> 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
        :param meta: metadata associated with the deprecation reason, such as the circumstances in which censorship was
                     detected
        :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()
        if reason not in [d.reason for d in self.deprecations]:
            new_deprecation = Deprecation(
                resource_type=type(self).__name__,
                resource_id=self.id,
                reason=reason,
                meta=meta
            )
            db.session.add(new_deprecation)
            return True
        logging.info("Not deprecating %s (reason=%s) because it's already deprecated with that reason.",
                     self.brn, reason)
        return False

    @property
    def deprecations(self) -> List[Deprecation]:
        return Deprecation.query.filter_by(  # type: ignore[no-any-return]
            resource_type='Proxy',
            resource_id=self.id
        ).all()

    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()
        ]