Models package

Here as a diagram of the database followed by their corresponding classes. I think they are simple enough to understand directly ;)

Database UML Diagram

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)