majuna/app/models/__init__.py

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

from sqlalchemy.orm import Mapped, mapped_column

from app.brm.brn import BRN
from app.extensions import db
from app.models.types import AwareDateTime


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

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

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

    def destroy(self) -> None:
        self.destroyed = datetime.now(tz=timezone.utc)
        self.updated = datetime.now(tz=timezone.utc)

    @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]
    resource_id: Mapped[int]
    deprecated_at: Mapped[datetime] = mapped_column(AwareDateTime())
    meta: Mapped[Optional[Dict[str, Any]]] = mapped_column(db.JSON())
    reason: Mapped[str]

    @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(AwareDateTime())
    updated: Mapped[datetime] = mapped_column(AwareDateTime())
    deprecated: Mapped[Optional[datetime]] = mapped_column(
        AwareDateTime(), nullable=True
    )
    deprecation_reason: Mapped[Optional[str]]
    destroyed: Mapped[Optional[datetime]] = mapped_column(
        AwareDateTime(), 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:
        if added is None:
            added = datetime.now(tz=timezone.utc)
        if updated is None:
            updated = datetime.now(tz=timezone.utc)
        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.now(tz=timezone.utc)
            self.deprecation_reason = reason
            self.updated = datetime.now(tz=timezone.utc)
        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.now(tz=timezone.utc)
        self.updated = datetime.now(tz=timezone.utc)

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