from .util import AttrCheck from sqlalchemy import Column, String, Integer, Computed, Boolean, DateTime from sqlalchemy import ForeignKey, UniqueConstraint, CheckConstraint from sqlalchemy import column, cast from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() class TException(Base, AttrCheck): __tablename__ = "Exceptions" exception_id = Column("ExceptionId", Integer, primary_key=True) name = Column("Name", String, nullable=False) fail = Column("Fail", Boolean, nullable=False) # children: kinds = relationship("TKind", back_populates="exception") messages = relationship("TMessage", back_populates="exception") class TExecution(Base, AttrCheck): __tablename__ = "Executions" execution_id = Column("ExecutionId", Integer, primary_key=True) start_date = Column("StartDate", DateTime, nullable=False) finish_date = Column("FinishDate", DateTime) completed = Column("Completed", Boolean, default=0, nullable=False) # children: messages = relationship("TMessage", back_populates="execution") as_int = column("AsInt") class TAddress(Base, AttrCheck): __tablename__ = "Ips" __table_args__ = ( UniqueConstraint("AsInt"), CheckConstraint((as_int >= 0) & (as_int < 2**32)), ) address_id = Column("IpId", Integer, primary_key=True) str = Column("AsStr", String, Computed( cast(as_int.op(">>")(24).op("&")(255), String) + "." + cast(as_int.op(">>")(16).op("&")(255), String) + "." + cast(as_int.op(">>")(8).op("&")(255), String) + "." + cast(as_int.op("&")(255), String) ), nullable=False) ip = Column("AsInt", Integer, nullable=False) china = Column("China", Boolean, default=0, nullable=False) block_target = Column("BlockTarget", Boolean, default=0, nullable=False) server = Column("Server", Boolean, default=0, nullable=False) redirect_target = Column("RedirectTarget", Boolean, default=0, nullable=False) gfw_target = Column("GfwTarget", Boolean, default=0, nullable=False) country_code = Column("CountryCode", String) # children: messages = relationship("TMessage", back_populates="server") records = relationship("TRecord", back_populates="address") class TKind(Base, AttrCheck): __tablename__ = "Kinds" __table_args__ = ( UniqueConstraint("Name"), ) kind_id = Column("KindId", Integer, primary_key=True) name = Column("Name", String, nullable=False) xxid = Column("ExpectExceptionId", Integer, ForeignKey("Exceptions.ExceptionId")) # parents: exception = relationship("TException", back_populates="kinds") # children: domains = relationship("TDomain", back_populates="kind") class TDomain(Base, AttrCheck): __tablename__ = "Domains" __table_args__ = ( UniqueConstraint("Name"), ) domain_id = Column("DomainId", Integer, primary_key=True) name = Column("Name", String, nullable=False) kind_id = Column("KindId", Integer, ForeignKey("Kinds.KindId")) # parents: kind = relationship("TKind", back_populates="domains") # children: messages = relationship("TMessage", back_populates="domain") class TRecord(Base, AttrCheck): __tablename__ = "Records" row_id = Column("rowid", Integer, primary_key=True) record_id = Column("RecordId", Integer, nullable=False) address_id = Column("IpId", Integer, ForeignKey("Ips.IpId")) # parents: address = relationship("TAddress", back_populates="records") class TMessage(Base, AttrCheck): __tablename__ = "Messages" message_id = Column("MessageId", Integer, primary_key=True) execution_id = Column("ExecutionId", Integer, ForeignKey("Executions.ExecutionId")) server_id = Column("ServerId", Integer, ForeignKey("Ips.IpId"), nullable=False) domain_id = Column("DomainId", Integer, ForeignKey("Domains.DomainId"), nullable=False) record_id = Column("RecordId", Integer) exception_id = Column("ExceptionId", Integer, ForeignKey("Exceptions.ExceptionId")) failed = Column("Failed", Boolean, nullable=False) # parents: execution = relationship("TExecution", back_populates="messages") server = relationship("TAddress", back_populates="messages") domain = relationship("TDomain", back_populates="messages") exception = relationship("TException", back_populates="messages")