SQLite and JSON toggle support added
Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
Jason
@@ -0,0 +1,368 @@
|
||||
"""
|
||||
SQLite Data Access Layer
|
||||
Provides database operations using SQLAlchemy
|
||||
"""
|
||||
from models import db, User, Product, Location, ProductAttribute
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
# ============================================================================
|
||||
# USER OPERATIONS
|
||||
# ============================================================================
|
||||
|
||||
def get_user_by_username(username):
|
||||
"""Get a user by username"""
|
||||
return User.query.filter_by(username=username).first()
|
||||
|
||||
def get_user_by_id(user_id):
|
||||
"""Get a user by ID"""
|
||||
return User.query.get(user_id)
|
||||
|
||||
def get_all_users():
|
||||
"""Get all users"""
|
||||
users = User.query.all()
|
||||
return [user.to_dict() for user in users]
|
||||
|
||||
def create_user(user_data):
|
||||
"""
|
||||
Create a new user
|
||||
|
||||
Args:
|
||||
user_data (dict): User data with keys: username, password, defaultLocation,
|
||||
locationSettings, permissions, active
|
||||
|
||||
Returns:
|
||||
dict: Created user as dictionary or None if username exists
|
||||
"""
|
||||
# Check if username already exists
|
||||
if User.query.filter_by(username=user_data['username']).first():
|
||||
return None
|
||||
|
||||
# Hash password if not already hashed
|
||||
password = user_data['password']
|
||||
if not password.startswith('pbkdf2:'):
|
||||
password = generate_password_hash(password, method='pbkdf2:sha256')
|
||||
|
||||
# Get location settings
|
||||
location_settings = user_data.get('locationSettings', {})
|
||||
default_location = user_data['defaultLocation']
|
||||
|
||||
# Ensure default location is accessible
|
||||
if default_location not in location_settings:
|
||||
location_settings[default_location] = {}
|
||||
location_settings[default_location]['accessible'] = True
|
||||
|
||||
# Create user
|
||||
user = User(
|
||||
username=user_data['username'],
|
||||
password=password,
|
||||
default_location=default_location,
|
||||
active=user_data.get('active', True),
|
||||
super_admin=user_data.get('superAdmin', False)
|
||||
)
|
||||
user.locationSettings = location_settings
|
||||
user.permissions = user_data.get('permissions', {})
|
||||
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
|
||||
return user.to_dict()
|
||||
|
||||
def update_user(user_id, user_data):
|
||||
"""
|
||||
Update an existing user
|
||||
|
||||
Args:
|
||||
user_id (int): User ID
|
||||
user_data (dict): Updated user data
|
||||
|
||||
Returns:
|
||||
dict: Updated user as dictionary or None if not found
|
||||
"""
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
# Update fields
|
||||
if 'defaultLocation' in user_data:
|
||||
user.default_location = user_data['defaultLocation']
|
||||
if 'active' in user_data:
|
||||
user.active = user_data['active']
|
||||
if 'superAdmin' in user_data:
|
||||
user.super_admin = user_data['superAdmin']
|
||||
if 'locationSettings' in user_data:
|
||||
user.locationSettings = user_data['locationSettings']
|
||||
if 'permissions' in user_data:
|
||||
user.permissions = user_data['permissions']
|
||||
if 'password' in user_data:
|
||||
password = user_data['password']
|
||||
if not password.startswith('pbkdf2:'):
|
||||
password = generate_password_hash(password, method='pbkdf2:sha256')
|
||||
user.password = password
|
||||
|
||||
db.session.commit()
|
||||
return user.to_dict()
|
||||
|
||||
def delete_user(user_id):
|
||||
"""
|
||||
Delete a user
|
||||
|
||||
Args:
|
||||
user_id (int): User ID
|
||||
|
||||
Returns:
|
||||
bool: True if deleted, False if not found
|
||||
"""
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return False
|
||||
|
||||
db.session.delete(user)
|
||||
db.session.commit()
|
||||
return True
|
||||
|
||||
def toggle_user_active(user_id, active):
|
||||
"""
|
||||
Toggle user active status
|
||||
|
||||
Args:
|
||||
user_id (int): User ID
|
||||
active (bool): Active status
|
||||
|
||||
Returns:
|
||||
dict: Updated user or None if not found
|
||||
"""
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
user.active = active
|
||||
db.session.commit()
|
||||
return user.to_dict()
|
||||
|
||||
def change_user_password(user_id, new_password):
|
||||
"""
|
||||
Change a user's password
|
||||
|
||||
Args:
|
||||
user_id (int): User ID
|
||||
new_password (str): New password (will be hashed)
|
||||
|
||||
Returns:
|
||||
dict: Updated user or None if not found
|
||||
"""
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
hashed_password = generate_password_hash(new_password, method='pbkdf2:sha256')
|
||||
user.password = hashed_password
|
||||
db.session.commit()
|
||||
return user.to_dict()
|
||||
|
||||
# ============================================================================
|
||||
# PRODUCT OPERATIONS
|
||||
# ============================================================================
|
||||
|
||||
def get_all_products(location=None):
|
||||
"""
|
||||
Get all products, optionally filtered by location
|
||||
|
||||
Args:
|
||||
location (str, optional): Filter by location code
|
||||
|
||||
Returns:
|
||||
list: List of product dictionaries
|
||||
"""
|
||||
query = Product.query
|
||||
if location:
|
||||
query = query.filter_by(location=location)
|
||||
|
||||
products = query.all()
|
||||
return [product.to_dict() for product in products]
|
||||
|
||||
def get_product_by_code(product_code):
|
||||
"""
|
||||
Get a product by its product code
|
||||
|
||||
Args:
|
||||
product_code (str): Product code
|
||||
|
||||
Returns:
|
||||
dict: Product dictionary or None if not found
|
||||
"""
|
||||
product = Product.query.filter_by(product_code=product_code).first()
|
||||
return product.to_dict() if product else None
|
||||
|
||||
def create_product(product_data):
|
||||
"""
|
||||
Create a new product
|
||||
|
||||
Args:
|
||||
product_data (dict): Product data
|
||||
|
||||
Returns:
|
||||
dict: Created product or None if code exists
|
||||
"""
|
||||
# Check if product already exists
|
||||
if Product.query.filter_by(product_code=product_data['productCode']).first():
|
||||
return None
|
||||
|
||||
# Create product
|
||||
product = Product(
|
||||
product_code=product_data['productCode'],
|
||||
category=product_data.get('category'),
|
||||
description=product_data.get('description'),
|
||||
discontinued=product_data.get('discontinued', False),
|
||||
location=product_data.get('location'),
|
||||
base_type=product_data.get('baseType', ''),
|
||||
is_accessory=product_data.get('isAccessory', False)
|
||||
)
|
||||
|
||||
# Set JSON properties
|
||||
product.subType = product_data.get('subType', {})
|
||||
product.materials = product_data.get('materials', [])
|
||||
product.colors = product_data.get('colors', [])
|
||||
product.compatibleAccessories = product_data.get('compatibleAccessories', [])
|
||||
product.imageConfig = product_data.get('imageConfig', {})
|
||||
|
||||
db.session.add(product)
|
||||
db.session.commit()
|
||||
|
||||
return product.to_dict()
|
||||
|
||||
def update_product(product_code, product_data):
|
||||
"""
|
||||
Update an existing product
|
||||
|
||||
Args:
|
||||
product_code (str): Product code
|
||||
product_data (dict): Updated product data
|
||||
|
||||
Returns:
|
||||
dict: Updated product or None if not found
|
||||
"""
|
||||
product = Product.query.filter_by(product_code=product_code).first()
|
||||
if not product:
|
||||
return None
|
||||
|
||||
# Update fields
|
||||
if 'category' in product_data:
|
||||
product.category = product_data['category']
|
||||
if 'description' in product_data:
|
||||
product.description = product_data['description']
|
||||
if 'discontinued' in product_data:
|
||||
product.discontinued = product_data['discontinued']
|
||||
if 'location' in product_data:
|
||||
product.location = product_data['location']
|
||||
if 'baseType' in product_data:
|
||||
product.base_type = product_data['baseType']
|
||||
if 'isAccessory' in product_data:
|
||||
product.is_accessory = product_data['isAccessory']
|
||||
|
||||
# Update JSON properties
|
||||
if 'subType' in product_data:
|
||||
product.subType = product_data['subType']
|
||||
if 'materials' in product_data:
|
||||
product.materials = product_data['materials']
|
||||
if 'colors' in product_data:
|
||||
product.colors = product_data['colors']
|
||||
if 'compatibleAccessories' in product_data:
|
||||
product.compatibleAccessories = product_data['compatibleAccessories']
|
||||
if 'imageConfig' in product_data:
|
||||
product.imageConfig = product_data['imageConfig']
|
||||
|
||||
db.session.commit()
|
||||
return product.to_dict()
|
||||
|
||||
def delete_product(product_code):
|
||||
"""
|
||||
Delete a product
|
||||
|
||||
Args:
|
||||
product_code (str): Product code
|
||||
|
||||
Returns:
|
||||
bool: True if deleted, False if not found
|
||||
"""
|
||||
product = Product.query.filter_by(product_code=product_code).first()
|
||||
if not product:
|
||||
return False
|
||||
|
||||
db.session.delete(product)
|
||||
db.session.commit()
|
||||
return True
|
||||
|
||||
# ============================================================================
|
||||
# LOCATION OPERATIONS
|
||||
# ============================================================================
|
||||
|
||||
def get_all_locations():
|
||||
"""Get all locations"""
|
||||
locations = Location.query.filter_by(active=True).all()
|
||||
return [loc.to_dict() for loc in locations]
|
||||
|
||||
def get_location_by_code(code):
|
||||
"""Get a location by code"""
|
||||
location = Location.query.filter_by(code=code).first()
|
||||
return location.to_dict() if location else None
|
||||
|
||||
# ============================================================================
|
||||
# PRODUCT ATTRIBUTES OPERATIONS
|
||||
# ============================================================================
|
||||
|
||||
def get_product_attributes():
|
||||
"""
|
||||
Get product attributes configuration
|
||||
|
||||
Returns:
|
||||
dict: Attributes with locations, statuses, productTypes, codeModifiers
|
||||
"""
|
||||
attributes = ProductAttribute.query.first()
|
||||
|
||||
if not attributes:
|
||||
return {
|
||||
'locations': get_all_locations(),
|
||||
'statuses': [],
|
||||
'productTypes': [],
|
||||
'codeModifiers': []
|
||||
}
|
||||
|
||||
return attributes.to_dict()
|
||||
|
||||
def update_product_attributes(attributes_data):
|
||||
"""
|
||||
Update product attributes configuration
|
||||
|
||||
Args:
|
||||
attributes_data (dict): Updated attributes
|
||||
|
||||
Returns:
|
||||
dict: Updated attributes
|
||||
"""
|
||||
attributes = ProductAttribute.query.first()
|
||||
|
||||
if not attributes:
|
||||
attributes = ProductAttribute()
|
||||
db.session.add(attributes)
|
||||
|
||||
if 'statuses' in attributes_data:
|
||||
attributes.statuses = attributes_data['statuses']
|
||||
if 'productTypes' in attributes_data:
|
||||
attributes.productTypes = attributes_data['productTypes']
|
||||
if 'codeModifiers' in attributes_data:
|
||||
attributes.codeModifiers = attributes_data['codeModifiers']
|
||||
|
||||
db.session.commit()
|
||||
return attributes.to_dict()
|
||||
|
||||
# ============================================================================
|
||||
# UTILITY FUNCTIONS
|
||||
# ============================================================================
|
||||
|
||||
def commit():
|
||||
"""Commit current transaction"""
|
||||
db.session.commit()
|
||||
|
||||
def rollback():
|
||||
"""Rollback current transaction"""
|
||||
db.session.rollback()
|
||||
Reference in New Issue
Block a user