import bcrypt from sqlalchemy import ( Column, Integer, BigInteger, Numeric, String, ForeignKey, DateTime, Boolean, Enum, ) from sqlalchemy.orm import relationship from sqlalchemy.sql import expression from sqlalchemy.ext.compiler import compiles from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.types import DateTime as DatetimeType BCRYPT_LOG_ROUNDS = 11 Base = declarative_base() class utcnow(expression.FunctionElement): type = DatetimeType() @compiles(utcnow, "postgresql") def pg_utcnow(element, compiler, **kw): return "TIMEZONE('utc', CURRENT_TIMESTAMP)" class DBMixin: id = Column(BigInteger, primary_key=True) created_date = Column(DateTime(timezone=True), server_default=utcnow()) modified_date = Column( DateTime(timezone=True), server_default=utcnow(), onupdate=utcnow() ) class UserModel(DBMixin, Base): __tablename__ = "user" email = Column(String(255), nullable=False, unique=True) password = Column(String(255)) geography = relationship("GeographyModel", back_populates="user") admin = Column(Boolean, nullable=False, default=False) confirmed = Column(Boolean, nullable=False, default=False) active = Column(Boolean, nullable=False, default=True) def __init__(self, email, password, admin=False): self.email = email self.password = bcrypt.hashpw( password.encode("utf-8"), bcrypt.gensalt(BCRYPT_LOG_ROUNDS) ).decode() self.admin = admin def check_password(self, password): return bcrypt.checkpw(password.encode("utf-8"), self.password.encode("utf-8")) class GeographyModel(Base, DBMixin): __tablename__ = "geography" name = Column(String(255)) user_id = Column(BigInteger, ForeignKey("user.id"), nullable=True) active = Column(Boolean, nullable=False, default=True) user = relationship("UserModel", uselist=False, back_populates="geography") stores = relationship( "StoreModel", order_by="StoreModel.id", back_populates="geography" ) class StoreModel(Base, DBMixin): __tablename__ = "store" geography_id = Column(BigInteger, ForeignKey("geography.id"), nullable=True) name = Column(String(255)) number = Column(String(4), nullable=True) address = Column(String(255)) city = Column(String(255)) state = Column(String(2)) zip = Column(String(12)) lat = Column(Numeric(10, 7), nullable=True) long = Column(Numeric(10, 7), nullable=True) tdlinx = Column(String(8), nullable=True) active = Column(Boolean, nullable=False, default=True) geography = relationship("GeographyModel", back_populates="stores") alerts = relationship("AlertModel", back_populates="store") class AlertModel(Base, DBMixin): __tablename__ = "alert" store_id = Column(BigInteger, ForeignKey("store.id")) store = relationship("StoreModel", back_populates="alerts") promo_name = Column(String(255)) response = Column( Enum("", "manager_refused", "valid", "invalid", name="response_type") ) responded = Column(Boolean, default=False) valid = Column(Boolean, nullable=True) active = Column(Boolean, nullable=False, default=True)