Models package¶
Here as a diagram of the database followed by their corresponding classes. I think they are simple enough to understand directly ;)
Redirect Rule¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | from datetime import datetime
from kubi_ecs_logger import Logger, Severity
from sqlalchemy import Column, Integer, DateTime, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class RedirectRule(base):
__tablename__ = "redirect_rule"
id = Column(Integer, autoincrement=True, primary_key=True)
domain_rule_id = Column(Integer, ForeignKey('domain_rule.id'), nullable=False)
domain_rule = relationship("DomainRule", lazy="joined", foreign_keys=[domain_rule_id])
path_rule_id = Column(Integer, ForeignKey("path_rule.id"), nullable=False)
path_rule = relationship("PathRule", lazy="joined", foreign_keys=[path_rule_id])
destination_rule_id = Column(Integer, ForeignKey("destination_rule.id"), nullable=False)
destination_rule = relationship("DestinationRule", lazy="joined", foreign_keys=[destination_rule_id])
weight = Column(Integer, nullable=False, default=100)
created_at = Column(DateTime, default=datetime.now())
modified_at = Column(DateTime, default=datetime.now())
__table_args__ = (UniqueConstraint('domain_rule_id', 'path_rule_id', 'destination_rule_id',
name='_domain_path_destination_uc'),)
def modify(self):
self.modified_at = datetime.now()
def delete(self, db_session, safe: bool = True):
db_session.delete(self)
db_session.commit()
Logger() \
.event(category="database", action="redirect rule deleted") \
.log(original=f"Redirect rule with id: {self.id} has been deleted") \
.out(severity=Severity.DEBUG)
self.domain_rule.delete(db_session, safe=safe)
self.path_rule.delete(db_session, safe=safe)
self.destination_rule.delete(db_session, safe=safe)
|
Path Rule¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | from datetime import datetime
from kubi_ecs_logger import Logger, Severity
from sqlalchemy import Column, Integer, DateTime, String, Boolean, UniqueConstraint, select, func
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class PathRule(base):
__tablename__ = "path_rule"
id = Column(Integer, autoincrement=True, primary_key=True)
rule = Column(String(1000))
is_regex = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.now())
modified_at = Column(DateTime, default=datetime.now())
__table_args__ = (UniqueConstraint("rule", "is_regex", name="_rule_regex_uc"),)
def modify(self):
self.modified_at = datetime.now()
def delete(self, db_session, safe: bool = True):
if safe:
from redirectory.libs_int.database import get_usage_count
if get_usage_count(db_session, type(self), self.id) > 0:
return
db_session.delete(self)
db_session.commit()
Logger() \
.event(category="database", action="path deleted") \
.log(original=f"Path with id: {self.id} has been deleted") \
.out(severity=Severity.DEBUG)
|
Domain Rule¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | from datetime import datetime
from kubi_ecs_logger import Logger, Severity
from sqlalchemy import Column, Integer, DateTime, String, Boolean, UniqueConstraint, select, func
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class DomainRule(base):
__tablename__ = "domain_rule"
id = Column(Integer, autoincrement=True, primary_key=True)
rule = Column(String(1000))
is_regex = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.now())
modified_at = Column(DateTime, default=datetime.now())
__table_args__ = (UniqueConstraint("rule", "is_regex", name="_rule_regex_uc"),)
def modify(self):
self.modified_at = datetime.now()
def delete(self, db_session, safe: bool = True):
if safe:
from redirectory.libs_int.database import get_usage_count
if get_usage_count(db_session, type(self), self.id) > 0:
return
db_session.delete(self)
db_session.commit()
Logger() \
.event(category="database", action="domain deleted") \
.log(original=f"Domain with id: {self.id} has been deleted") \
.out(severity=Severity.DEBUG)
|
Destination Rule¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | from datetime import datetime
from kubi_ecs_logger import Logger, Severity
from sqlalchemy import Column, Integer, DateTime, String, Boolean, UniqueConstraint, select, func
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class DestinationRule(base):
__tablename__ = "destination_rule"
id = Column(Integer, autoincrement=True, primary_key=True)
destination_url = Column(String(1000))
is_rewrite = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.now())
modified_at = Column(DateTime, default=datetime.now())
__table_args__ = (UniqueConstraint("destination_url", "is_rewrite", name="_destination_rewrite_uc"),)
def modify(self):
self.modified_at = datetime.now()
def delete(self, db_session, safe: bool = True):
if safe:
from redirectory.libs_int.database import get_usage_count
if get_usage_count(db_session, type(self), self.id) > 0:
return
db_session.delete(self)
db_session.commit()
Logger() \
.event(category="database", action="destination deleted") \
.log(original=f"Destination with id: {self.id} has been deleted") \
.out(severity=Severity.DEBUG)
|
Ambiguous Requests¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | from datetime import datetime
from sqlalchemy import Column, Integer, DateTime, String
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class AmbiguousRequest(base):
__tablename__ = "ambiguous_request"
id = Column(Integer, autoincrement=True, primary_key=True)
request = Column(String(1000), unique=True)
created_at = Column(DateTime, default=datetime.now())
|
Hyperscan DB Version¶
1 2 3 4 5 6 7 8 9 10 11 12 13 | from sqlalchemy import Column, Integer, String
from redirectory.libs_int.database import DatabaseManager
base = DatabaseManager().get_base()
class HsDbVersion(base):
__tablename__ = "hs_db_version"
id = Column(Integer, autoincrement=True, primary_key=True)
old_version = Column(String, nullable=True)
current_version = Column(String, nullable=False)
|