95 lines
4.3 KiB
Python
95 lines
4.3 KiB
Python
import uuid
|
|
import os
|
|
|
|
import dateutil.relativedelta
|
|
from datetime import datetime
|
|
from sqlalchemy import Enum, UniqueConstraint
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
|
|
from loguru import logger
|
|
from overseer.flask import db, SQLITE_MODE
|
|
|
|
uuid_column_type = lambda: UUID(as_uuid=True) if not SQLITE_MODE else db.String(36)
|
|
|
|
|
|
class Guarantee(db.Model):
|
|
__tablename__ = "guarantees"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
guarantor_id = db.Column(db.Integer, db.ForeignKey("instances.id", ondelete="CASCADE"), nullable=False)
|
|
guarantor_instance = db.relationship("Instance", back_populates="guarantees", foreign_keys=[guarantor_id])
|
|
guaranteed_id = db.Column(db.Integer, db.ForeignKey("instances.id", ondelete="CASCADE"), unique=True, nullable=False)
|
|
guaranteed_instance = db.relationship("Instance", back_populates="guarantors", foreign_keys=[guaranteed_id])
|
|
created = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
|
|
|
|
|
|
class Endorsement(db.Model):
|
|
__tablename__ = "endorsements"
|
|
__table_args__ = (UniqueConstraint('approving_id', 'endorsed_id', name='endoresements_approving_id_endorsed_id'),)
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
approving_id = db.Column(db.Integer, db.ForeignKey("instances.id", ondelete="CASCADE"), nullable=False)
|
|
approving_instance = db.relationship("Instance", back_populates="approvals", foreign_keys=[approving_id])
|
|
endorsed_id = db.Column(db.Integer, db.ForeignKey("instances.id", ondelete="CASCADE"), nullable=False)
|
|
endorsed_instance = db.relationship("Instance", back_populates="endorsements", foreign_keys=[endorsed_id])
|
|
created = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
|
|
|
|
|
|
class Instance(db.Model):
|
|
__tablename__ = "instances"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
domain = db.Column(db.String(255), unique=True, nullable=False, index=True)
|
|
api_key = db.Column(db.String(100), unique=True, nullable=False, index=True)
|
|
created = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
|
|
updated = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
|
|
oprhan_since = db.Column(db.DateTime, nullable=True)
|
|
|
|
open_registrations = db.Column(db.Boolean, unique=False, nullable=False, index=True)
|
|
email_verify = db.Column(db.Boolean, unique=False, nullable=False, index=True)
|
|
software = db.Column(db.String(50), unique=False, nullable=False, index=True)
|
|
|
|
approvals = db.relationship("Endorsement", back_populates="approving_instance", cascade="all, delete-orphan", foreign_keys=[Endorsement.approving_id])
|
|
endorsements = db.relationship("Endorsement", back_populates="endorsed_instance", cascade="all, delete-orphan", foreign_keys=[Endorsement.endorsed_id])
|
|
guarantees = db.relationship("Guarantee", back_populates="guarantor_instance", cascade="all, delete-orphan", foreign_keys=[Guarantee.guarantor_id])
|
|
guarantors = db.relationship("Guarantee", back_populates="guaranteed_instance", cascade="all, delete-orphan", foreign_keys=[Guarantee.guaranteed_id])
|
|
|
|
def create(self):
|
|
db.session.add(self)
|
|
db.session.commit()
|
|
|
|
def get_details(self):
|
|
ret_dict = {
|
|
"id": self.id,
|
|
"domain": self.domain,
|
|
"open_registrations": self.open_registrations,
|
|
"email_verify": self.email_verify,
|
|
"endorsements": len(self.endorsements),
|
|
"approvals": len(self.approvals),
|
|
"guarantor": self.get_guarantor_domain(),
|
|
}
|
|
return ret_dict
|
|
|
|
|
|
def get_guarantee(self):
|
|
if len(self.guarantors) == 0:
|
|
return None
|
|
return self.guarantors[0]
|
|
|
|
def get_guarantor(self):
|
|
guarantee = self.get_guarantee()
|
|
if not guarantee:
|
|
return None
|
|
return guarantee.guarantor_instance
|
|
return Instance.query.filter_by(id=guarantee.guarantor_id).first()
|
|
|
|
def get_guarantor_domain(self):
|
|
guarantor = self.get_guarantor()
|
|
return guarantor.domain if guarantor else None
|
|
|
|
def set_as_oprhan(self):
|
|
self.oprhan_since = datetime.utcnow()
|
|
db.session.commit()
|
|
|
|
def unset_as_orphan(self):
|
|
self.oprhan_since = None
|
|
db.session.commit()
|
|
|