Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- @declarative_mixin
- class EntityRefMixin:
- entity_id = db.Column(db.Integer, nullable=False)
- entity_type = db.Column(
- db.Enum("reporter", "account", "report", "job", name="entity_type_enum"),
- nullable=False
- )
- @property
- def entity(self):
- return getattr(self, self.entity_type)
- @declarative_mixin
- class EntityRelationshipMixin:
- @classmethod
- def __declare_last__(cls):
- """
- Use __tablename__ to populate entity_type at time of creation.
- Function gets called after mappings are completed.
- https://docs.sqlalchemy.org/en/14/orm/declarative_config.html#declare-last
- """
- event.listen(cls.gs, "set", cls.set_entity_type, retval=False)
- event.listen(cls.bq, "set", cls.set_entity_type, retval=False)
- event.listen(cls.reporter_fields, "append", cls.append_entity_type, retval=False)
- @classmethod
- def set_entity_type(cls, target, new_value, old_value, initiator):
- new_value.entity_type = cls.__tablename__
- @classmethod
- def append_entity_type(cls, state, value, initiator):
- value.entity_type = cls.__tablename__
- @classmethod
- def join_builder(cls, remote_cls):
- return and_(
- foreign(remote_cls.entity_type) == cls.__tablename__,
- foreign(remote_cls.entity_id) == cls.id
- )
- # region relationships
- @declared_attr
- def gs(cls): # noqa
- return db.relationship(
- "GoogleStorage",
- primaryjoin=lambda: cls.join_builder(GoogleStorage),
- backref=db.backref(cls.__tablename__, uselist=False, overlaps="reporter,account"),
- overlaps=f"reporter,account,gs,{cls.__tablename__}",
- uselist=False
- )
- @declared_attr
- def bq(cls): # noqa
- return db.relationship(
- "BigQuery",
- primaryjoin=lambda: cls.join_builder(BigQuery),
- backref=db.backref(cls.__tablename__, uselist=False, overlaps="reporter,account"),
- overlaps=f"reporter,account,bq,{cls.__tablename__}",
- uselist=False
- )
- @declared_attr
- def reporter_fields(cls): # noqa
- return db.relationship(
- "ParameterValue",
- primaryjoin=lambda: cls.join_builder(ParameterValue),
- lazy="dynamic",
- backref=db.backref(cls.__tablename__, overlaps="reporter,account"),
- overlaps=f"reporter,account,reporter_fields,{cls.__tablename__}"
- )
- # endregion
- class GoogleStorage(EntityRefMixin, db.Model):
- __tablename__ = "google_storage"
- id = db.Column(db.Integer, primary_key=True)
- use_gs = db.Column(db.Boolean, default=False)
- bucket_name = db.Column(db.String(50), nullable=True)
- folder_path = db.Column(db.String(100), nullable=True)
- class BigQuery(EntityRefMixin, db.Model):
- __tablename__ = "google_bigquery"
- id = db.Column(db.Integer, primary_key=True)
- use_bq = db.Column(db.Boolean, default=False)
- # bq_project = db.Column(db.String(50), nullable=True) # future feature
- dataset = db.Column(db.String(50), nullable=True)
- table = db.Column(db.String(50), nullable=True)
- columns = db.Column(MutableList.as_mutable(db.JSON), nullable=False, default=list())
- ingestion_action = db.Column(
- db.Enum("WRITE_TRUNCATE", "WRITE_APPEND", name="bq_ingestion_action_enum"),
- nullable=False,
- default="WRITE_APPEND"
- )
- class ParameterValue(EntityRefMixin, db.Model):
- """Assigns a value to a parameter and assigns the parameter to an entity."""
- __tablename__ = "parameter_value"
- __table_args__ = (
- db.UniqueConstraint(
- "entity_id", "entity_type", "parameter_id",
- name="entity_parameter_uc"
- ),
- )
- id = db.Column(db.Integer, primary_key=True)
- parameter_id = db.Column(db.Integer, db.ForeignKey("parameter.id"), nullable=False)
- str_value = db.Column(db.String(100), nullable=True)
- int_value = db.Column(db.Integer, nullable=True)
- bool_value = db.Column(db.Boolean, nullable=True)
- list_value = db.Column(MutableList.as_mutable(db.JSON), nullable=True)
- def __init__(self, **kwargs):
- """Explicit __init__ added to accommodate hybrid_property."""
- self.parameter = kwargs.pop("parameter")
- for kwarg, value in kwargs.items():
- setattr(self, kwarg, value)
- # region value hybrid_property
- @hybrid_property
- def value(self):
- return getattr(self, f"{self.parameter.type}_value")
- @value.setter # noqa
- def value(self, new_value):
- setattr(self, f"{self.parameter.type}_value", new_value)
- @value.deleter
- def value(self):
- # self._set_value(None)
- setattr(self, f"{self.parameter.type}_value", None)
- # endregion
- class Parameter(db.Model):
- __tablename__ = "parameter"
- id = db.Column(db.Integer, primary_key=True)
- # universal parameters, like skip_normalization and skip_archival fields, will not be assigned to a reporter
- reporter_id = db.Column(db.Integer, db.ForeignKey("reporter.id"), nullable=True)
- name = db.Column(db.String(45), nullable=False)
- type = db.Column(
- db.Enum("str", "int", "bool", "list", name="parameter_type_enum"),
- nullable=False,
- default="str"
- )
- display_name = db.Column(db.String(100), nullable=False) # nullable=True, default=None)
- is_required = db.Column(db.Boolean, default=False)
- help_text = db.Column(db.String(100), nullable=True, default=None)
- values = db.relationship("ParameterValue", backref="parameter")
- class Reporter(EntityRelationshipMixin, db.Model):
- __tablename__ = "reporter"
- # region internally-used fields
- id = db.Column(db.Integer, primary_key=True)
- module = db.Column(db.String(50), nullable=False)
- klass = db.Column(db.String(50), nullable=False)
- # endregion
- # region meta fields
- data_source = db.Column(db.String(50), nullable=True)
- display_name = db.Column(db.String(50), nullable=False, unique=True)
- # endregion
- # region behaviors
- omit_credentials = db.Column(db.Boolean, default=False)
- requires_otp = db.Column(db.Boolean, nullable=False, default=False)
- # endregion
- # region relationships
- # for managing the parameters to show for each Reporter
- parameters = db.relationship("Parameter", backref="reporter", lazy="dynamic", cascade="all, delete")
- accounts = db.relationship("Account", backref="reporter", cascade="all, delete")
- # endregion
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement