Gan, Jimmy
6c0d5a4e63
- Flask-based mold cost calculation application - PostgreSQL database with SQLAlchemy ORM - Docker/Podman containerization - Comprehensive documentation in docs/ - Clean, organized codebase without obsolete files - Production-ready deployment configuration - Enhanced pytest configuration with coverage reporting - Master/Development branch structure
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from flask_login import UserMixin
|
|
from sqlalchemy import event, func
|
|
from .. import db
|
|
import secrets
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class User(UserMixin, db.Model):
|
|
__tablename__ = 'enterprise_users'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(25), unique=True, nullable=False)
|
|
email = db.Column(db.String(120), unique=True, nullable=False)
|
|
company = db.Column(db.String(100), nullable=False)
|
|
password_hash = db.Column(db.String(128), nullable=False)
|
|
created_at = db.Column(db.DateTime)
|
|
password_changed_at = db.Column(db.DateTime)
|
|
is_admin = db.Column(db.Boolean, default=False)
|
|
last_login = db.Column(db.DateTime)
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
failed_login_attempts = db.Column(db.Integer, default=0)
|
|
last_failed_login = db.Column(db.DateTime)
|
|
reset_token = db.Column(db.String(100), unique=True)
|
|
reset_token_expires = db.Column(db.DateTime)
|
|
|
|
# Relationships
|
|
quotations = db.relationship('Quotation', backref='user', lazy=True)
|
|
|
|
def set_password(self, password):
|
|
try:
|
|
self.password_hash = generate_password_hash(password, method='pbkdf2:sha256')
|
|
self.password_changed_at = datetime.now(timezone.utc)
|
|
self.reset_token = None
|
|
self.reset_token_expires = None
|
|
except Exception as e:
|
|
logger.error(f'Error setting password: {str(e)}', exc_info=True)
|
|
raise
|
|
|
|
def check_password(self, password):
|
|
return check_password_hash(self.password_hash, password)
|
|
|
|
def generate_reset_token(self):
|
|
token = secrets.token_urlsafe(16) # Shorter token for display
|
|
self.reset_token = token
|
|
self.reset_token_expires = datetime.now(timezone.utc) + timedelta(minutes=15) # 15-minute expiry
|
|
return token
|
|
|
|
def verify_reset_token(self, token):
|
|
if not self.reset_token or not self.reset_token_expires:
|
|
return False
|
|
if datetime.now(timezone.utc) > self.reset_token_expires:
|
|
return False
|
|
return secrets.compare_digest(self.reset_token, token)
|
|
|
|
@classmethod
|
|
def authenticate(cls, email, password):
|
|
user = cls.query.filter(func.lower(cls.email) == email.lower()).first()
|
|
if user and user.check_password(password):
|
|
if not user.is_active:
|
|
return None
|
|
user.failed_login_attempts = 0
|
|
user.last_failed_login = None
|
|
db.session.commit()
|
|
return user
|
|
if user:
|
|
user.failed_login_attempts = (user.failed_login_attempts or 0) + 1
|
|
user.last_failed_login = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
return None
|
|
|
|
@classmethod
|
|
def create(cls, username, email, company, password):
|
|
try:
|
|
logger.info(f'Checking if username {username} exists...')
|
|
if cls.query.filter(func.lower(cls.username) == username.lower()).first():
|
|
logger.warning(f'Username {username} already taken')
|
|
raise ValueError('Username already taken')
|
|
|
|
logger.info(f'Checking if email {email} exists...')
|
|
if cls.query.filter(func.lower(cls.email) == email.lower()).first():
|
|
logger.warning(f'Email {email} already registered')
|
|
raise ValueError('Email already registered')
|
|
|
|
logger.info('Creating new user...')
|
|
user = cls(
|
|
username=username,
|
|
email=email,
|
|
company=company,
|
|
is_active=True,
|
|
failed_login_attempts=0,
|
|
is_admin=False
|
|
)
|
|
|
|
logger.info('Setting password...')
|
|
user.set_password(password)
|
|
|
|
logger.info('Adding user to session...')
|
|
db.session.add(user)
|
|
|
|
logger.info('User created successfully')
|
|
return user
|
|
except Exception as e:
|
|
logger.error(f'Error creating user: {str(e)}', exc_info=True)
|
|
raise
|
|
|
|
def __repr__(self):
|
|
return f'<User {self.username}>'
|
|
|
|
# Event listeners for model changes
|
|
@event.listens_for(User, 'before_insert')
|
|
def set_timestamps(mapper, connection, target):
|
|
target.created_at = datetime.now(timezone.utc)
|
|
target.password_changed_at = datetime.now(timezone.utc) |