import bcrypt from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Boolean, Numeric from sqlalchemy.orm import relationship, backref from sqlalchemy.schema import CreateColumn from sqlalchemy.sql import expression from sqlalchemy.ext.compiler import compiles from sqlalchemy.types import DateTime as DateTimeType BCRYPT_LOG_ROUNDS = 11 # can be moved to models util? class utcnow(expression.FunctionElement): type = DateTimeType() # Can be moved to the models util? @compiles(utcnow, 'postgresql') def pg_utcnow(element, compiler, **kw): return "TIMEZONE('utc', CURRENT_TIMESTAMP)" @compiles(CreateColumn, 'postgresql') def use_identity(element, compiler, **kw): text = compiler.visit_create_column(element, **kw) text = text.replace("SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY") return text Base = declarative_base() class DBMixin: id = Column(Integer, primary_key=True, autoincrement=True) created_date = Column(DateTime(timezone=True), server_default=utcnow()) modified_date = Column(DateTime(timezone=True), server_default=utcnow(), onupdate=utcnow()) def to_dict(self): d = self.__dict__.copy() if '_sa_instance_state' in d: d.pop('_sa_instance_state') return d def ReferenceCol(tablename, nullable=False, **kw): return Column(ForeignKey('{}.id'.format(tablename)), nullable=nullable, **kw) class Cookie(Base, DBMixin): __tablename__ = 'cookies' name = Column(String(50), index=True) recipe_url = Column(String(255)) sku = Column(String(55)) qoh = Column(Integer) unit_cost = Column(Numeric(12, 2)) class User(DBMixin, Base): __tablename__ = 'users' email = Column(String(255), nullable=False, unique=True) password = Column(String(255)) 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')) def to_dict(self): d = self.__dict__.copy() if '_sa_instance_state' in d: d.pop('_sa_instance_state') d.pop('password') return d class Order(DBMixin, Base): __tablename__ = 'orders' user_id = ReferenceCol('users') shipped = Column(Boolean, default=False) user = relationship('User', backref=backref('orders')) class LineItem(DBMixin, Base): __tablename__ = 'line_items' order_id = ReferenceCol('orders') cookie_id = ReferenceCol('cookies') quantity = Column(Integer) extended_cost = Column(Numeric(12, 2)) order = relationship('Order', backref=backref('line_items')) cookie = relationship('Cookie', uselist=False)