145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import logging
|
|
from abc import abstractmethod
|
|
from datetime import datetime
|
|
from typing import Union, List, Optional, Any, Dict
|
|
|
|
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 Deprecation(db.Model): # type: ignore[name-defined,misc]
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
resource_type = db.Column(db.String(50))
|
|
resource_id = db.Column(db.Integer)
|
|
deprecated_at = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
|
|
meta = db.Column(db.JSON())
|
|
reason = db.Column(db.String(), nullable=False)
|
|
|
|
@property
|
|
def resource(self) -> "AbstractResource":
|
|
from app.models.mirrors import Proxy
|
|
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 = 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, 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()
|
|
]
|