From 0632aa22e198ca39a125a13ecc0179412ae80100 Mon Sep 17 00:00:00 2001 From: Jason <17367223+jsoltys@users.noreply.github.com> Date: Tue, 5 May 2026 16:06:52 -0500 Subject: [PATCH] SQLite and JSON toggle support added Co-authored-by: Copilot --- app/DATA_ACCESS_LAYER.md | 171 ++++++++++++++ app/app.py | 57 ++++- app/blueprints/auth.py | 60 ++--- app/blueprints/products.py | 115 +++++++++- app/blueprints/users.py | 122 +++------- app/config.py | 24 ++ app/data_access.py | 91 ++++++++ app/data_access_json.py | 449 +++++++++++++++++++++++++++++++++++++ app/data_access_sqlite.py | 368 ++++++++++++++++++++++++++++++ app/migrate_to_sqlite.py | 278 +++++++++++++++++++++++ app/models.py | 262 ++++++++++++++++++++++ app/requirements.txt | 1 + 12 files changed, 1862 insertions(+), 136 deletions(-) create mode 100644 app/DATA_ACCESS_LAYER.md create mode 100644 app/config.py create mode 100644 app/data_access.py create mode 100644 app/data_access_json.py create mode 100644 app/data_access_sqlite.py create mode 100644 app/migrate_to_sqlite.py create mode 100644 app/models.py diff --git a/app/DATA_ACCESS_LAYER.md b/app/DATA_ACCESS_LAYER.md new file mode 100644 index 0000000..828e6b5 --- /dev/null +++ b/app/DATA_ACCESS_LAYER.md @@ -0,0 +1,171 @@ +# Data Access Layer (DAL) - SQLite and JSON Toggle + +The application now supports switching between **SQLite** and **JSON** data storage through a simple configuration toggle. + +## Configuration + +Open `app/config.py` and set: + +```python +# Set to True for SQLite, False for JSON files +USE_DATABASE = True # or False +``` + +## Architecture + +### Modules + +1. **config.py** - Application configuration + - `USE_DATABASE` - Toggle between SQLite and JSON + - Database URI and other settings + +2. **data_access_sqlite.py** - SQLite implementation + - Uses SQLAlchemy models + - Supports transactions and rollback + +3. **data_access_json.py** - JSON file implementation + - Thread-safe file operations + - Immediate writes (no transactions) + +4. **data_access.py** - Universal wrapper + - Automatically imports correct module based on config + - Provides unified interface to all blueprints + +### Standardized Functions + +Both modules implement the same functions: + +**User Operations:** +- `get_user_by_username(username)` +- `get_user_by_id(user_id)` +- `get_all_users()` +- `create_user(user_data)` +- `update_user(user_id, user_data)` +- `delete_user(user_id)` +- `toggle_user_active(user_id, active)` +- `change_user_password(user_id, new_password)` + +**Product Operations:** +- `get_all_products(location=None)` +- `get_product_by_code(product_code)` +- `create_product(product_data)` +- `update_product(product_code, product_data)` +- `delete_product(product_code)` + +**Location Operations:** +- `get_all_locations()` +- `get_location_by_code(code)` + +**Product Attributes:** +- `get_product_attributes()` +- `update_product_attributes(attributes_data)` + +**Utilities:** +- `commit()` - Commit transaction (no-op for JSON) +- `rollback()` - Rollback transaction (no-op for JSON) + +## Usage in Blueprints + +All blueprints now import from the data access layer: + +```python +import data_access as da + +# Get user +user = da.get_user_by_username('Master') + +# Get products +products = da.get_all_products(location='IOLA') + +# Create product +product = da.create_product(product_data) +``` + +## Switching Between Modes + +### To Use SQLite: + +1. Set `USE_DATABASE = True` in `config.py` +2. Run migration: `python migrate_to_sqlite.py` +3. Start app: `python app.py` + +### To Use JSON: + +1. Set `USE_DATABASE = False` in `config.py` +2. Ensure JSON files exist in `data/` folder +3. Start app: `python app.py` + +## Key Differences + +| Feature | SQLite | JSON | +|---------|--------|------| +| **Transactions** | Yes | No (immediate writes) | +| **Concurrency** | Better | File locks | +| **Performance** | Faster for large datasets | Faster for small datasets | +| **Backup** | Single .db file | Multiple .json files | +| **Queries** | Complex SQL queries | Load entire file | +| **User IDs** | Database auto-increment | Array index | + +## Important Notes + +### User IDs +- **SQLite**: Uses database-generated IDs +- **JSON**: Uses array index (changes if users deleted) + +### Session Management +- Both modes use username-based sessions (not user_id) +- Session keys: `username`, `currentLocation`, `accessibleLocations` + +### Data Consistency +- **SQLite**: ACID compliant with rollback support +- **JSON**: No rollback - writes are immediate and final + +### File Serving +- The `/quiz/data/` route still serves static JSON files +- When using SQLite, these JSON files may become stale +- Consider using API endpoints instead for dynamic data + +## Migration + +To migrate existing JSON data to SQLite: + +```bash +cd app +python migrate_to_sqlite.py +``` + +This will: +- Create `data/products.db` +- Backup JSON files to `data/json_backup/` +- Migrate users, products, locations, and attributes +- Skip duplicates on subsequent runs + +## Testing + +Test both modes: + +```bash +# Test with SQLite +# Set USE_DATABASE = True in config.py +python app.py + +# Test with JSON +# Set USE_DATABASE = False in config.py +python app.py +``` + +## Troubleshooting + +**"Import 'flask' could not be resolved"** +- This is just an editor warning, not a runtime error +- Install dependencies: `pip install -r requirements.txt` + +**Data not updating** +- Check `USE_DATABASE` setting in config.py +- Verify correct data files exist (`.db` or `.json`) +- Check terminal for data access layer mode message + +**User not found after switching modes** +- User IDs differ between SQLite and JSON +- Clear browser cookies/session +- Re-login to create new session diff --git a/app/app.py b/app/app.py index 1ed1ab1..71af7af 100644 --- a/app/app.py +++ b/app/app.py @@ -1,6 +1,7 @@ from flask import Flask, session, redirect, url_for import secrets import os +import config # Get the directory where this script is located basedir = os.path.abspath(os.path.dirname(__file__)) @@ -12,8 +13,18 @@ app = Flask(__name__, # Configure session - generate a random secret key app.secret_key = secrets.token_hex(32) -app.config['SESSION_COOKIE_HTTPONLY'] = True -app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' +app.config['SESSION_COOKIE_HTTPONLY'] = config.SESSION_COOKIE_HTTPONLY +app.config['SESSION_COOKIE_SAMESITE'] = config.SESSION_COOKIE_SAMESITE + +# Configure SQLAlchemy only if using database +if config.USE_DATABASE: + app.config['SQLALCHEMY_DATABASE_URI'] = config.DATABASE_URI + app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = config.SQLALCHEMY_TRACK_MODIFICATIONS + app.config['SQLALCHEMY_ECHO'] = config.SQLALCHEMY_ECHO + + # Initialize database + from models import db + db.init_app(app) # Import and register blueprints from blueprints.auth import auth_bp @@ -82,6 +93,48 @@ def page_not_found(e): def internal_error(e): return "

500 - Internal Server Error

", 500 +# Database initialization route +@app.route('/init-db') +def init_db(): + """Initialize the database (create all tables)""" + if not config.USE_DATABASE: + return """ + + Database Not Enabled + +

⚠️ Database Not Enabled

+

The application is currently configured to use JSON files.

+

Set USE_DATABASE = True in config.py to enable SQLite.

+

Go to Home

+ + + """ + + try: + from models import db + with app.app_context(): + db.create_all() + return """ + + Database Initialized + +

✅ Database Initialized Successfully!

+

All tables have been created.

+

Go to Home

+ + + """ + except Exception as e: + return f""" + + Database Error + +

❌ Database Initialization Failed

+

Error: {str(e)}

+ + + """, 500 + # This is the critical line for Passenger application = app diff --git a/app/blueprints/auth.py b/app/blueprints/auth.py index db378d8..fdfb5d5 100644 --- a/app/blueprints/auth.py +++ b/app/blueprints/auth.py @@ -5,39 +5,21 @@ Handles login, logout, and session management from flask import Blueprint, render_template, request, jsonify, session, redirect, url_for from werkzeug.security import check_password_hash from functools import wraps -import json -import os +import data_access as da +import config auth_bp = Blueprint('auth', __name__) -# Path to users file -USERS_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data', 'users.json') - -def load_users(): - """Load users from JSON file""" - if os.path.exists(USERS_FILE): - try: - with open(USERS_FILE, 'r') as f: - return json.load(f) - except: - return [] - return [] - def login_required(f): """Decorator to require login for routes""" @wraps(f) def decorated_function(*args, **kwargs): - if 'user_id' not in session: + if 'username' not in session: return redirect(url_for('auth.login_page')) # Check if user is active - users = load_users() - if session['user_id'] >= len(users): - session.clear() - return redirect(url_for('auth.login_page', message='User not found')) - - user = users[session['user_id']] - if not user.get('active', True): + user = da.get_user_by_username(session['username']) + if not user or not user.get('active', True): session.clear() return redirect(url_for('auth.login_page', message='Account is inactive')) @@ -46,14 +28,10 @@ def login_required(f): def get_current_user(): """Get the currently logged in user""" - if 'user_id' not in session: + if 'username' not in session: return None - users = load_users() - if session['user_id'] >= len(users): - return None - - return users[session['user_id']] + return da.get_user_by_username(session['username']) def can_user(permission, location=None): """ @@ -98,14 +76,14 @@ def can_user(permission, location=None): @auth_bp.route('/login') def login_page(): """Serve the login page""" - if 'user_id' in session: + if 'username' in session: return redirect(url_for('home')) return render_template('login.html') @auth_bp.route('/select-location') def select_location_page(): """Serve the location selection page""" - if 'user_id' not in session: + if 'username' not in session: return redirect(url_for('auth.login_page')) return render_template('select_location.html') @@ -123,16 +101,8 @@ def login(): 'message': 'Username and password are required' }), 400 - users = load_users() - # Find user by username - user_index = None - user = None - for i, u in enumerate(users): - if u['username'] == username: - user_index = i - user = u - break + user = da.get_user_by_username(username) if not user: return jsonify({ @@ -155,14 +125,14 @@ def login(): }), 401 # Create session - session['user_id'] = user_index session['username'] = user['username'] session['defaultLocation'] = user.get('defaultLocation') # Get accessible locations accessible_locations = [] - if 'locationSettings' in user: - for loc_code, settings in user['locationSettings'].items(): + location_settings = user.get('locationSettings', {}) + if location_settings: + for loc_code, settings in location_settings.items(): if settings.get('accessible', False): accessible_locations.append(loc_code) @@ -190,7 +160,7 @@ def login(): @auth_bp.route('/api/select-location', methods=['POST']) def select_location(): """Select a location for the current session""" - if 'user_id' not in session: + if 'username' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' @@ -231,7 +201,7 @@ def logout(): @auth_bp.route('/api/session', methods=['GET']) def get_session(): """Get current session information""" - if 'user_id' not in session: + if 'username' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' diff --git a/app/blueprints/products.py b/app/blueprints/products.py index abbcd98..d2716dc 100644 --- a/app/blueprints/products.py +++ b/app/blueprints/products.py @@ -2,8 +2,9 @@ Product Finder Blueprint Handles the product quiz and recommendations """ -from flask import Blueprint, render_template, send_from_directory, session, jsonify, redirect, url_for +from flask import Blueprint, render_template, send_from_directory, session, jsonify, redirect, url_for, request from blueprints.auth import login_required, get_current_user, can_user +import data_access as da import os products_bp = Blueprint('products', __name__, url_prefix='/quiz') @@ -47,13 +48,123 @@ def get_user_session(): return jsonify({ 'username': session.get('username'), - 'userId': session.get('user_id'), + 'userId': None, # Not needed with username-based sessions 'currentLocation': session.get('currentLocation'), 'accessibleLocations': session.get('accessibleLocations', []), 'role': session.get('role', 'user'), 'permissions': permissions }) +# Product API Endpoints +@products_bp.route('/api/products', methods=['GET']) +@login_required +def get_products(): + """Get all products or filter by location""" + location = request.args.get('location') + products = da.get_all_products(location=location) + return jsonify({ + 'status': 'success', + 'products': products + }) + +@products_bp.route('/api/products/', methods=['GET']) +@login_required +def get_product(product_code): + """Get a specific product by code""" + product = da.get_product_by_code(product_code) + + if not product: + return jsonify({ + 'status': 'error', + 'message': 'Product not found' + }), 404 + + return jsonify({ + 'status': 'success', + 'product': product + }) + +@products_bp.route('/api/products', methods=['POST']) +@login_required +def create_product(): + """Create a new product""" + if not can_user('manage_products'): + return jsonify({ + 'status': 'error', + 'message': 'Permission denied' + }), 403 + + try: + data = request.json + + # Create product via data access layer + product = da.create_product(data) + + if not product: + return jsonify({ + 'status': 'error', + 'message': 'Product code already exists' + }), 400 + + return jsonify({ + 'status': 'success', + 'message': 'Product created successfully', + 'product': product + }) + + except Exception as e: + da.rollback() + return jsonify({ + 'status': 'error', + 'message': str(e) + }), 500 + +@products_bp.route('/api/products/', methods=['PUT']) +@login_required +def update_product(product_code): + """Update an existing product""" + if not can_user('manage_products'): + return jsonify({ + 'status': 'error', + 'message': 'Permission denied' + }), 403 + + try: + data = request.json + + # Update product via data access layer + product = da.update_product(product_code, data) + + if not product: + return jsonify({ + 'status': 'error', + 'message': 'Product not found' + }), 404 + + return jsonify({ + 'status': 'success', + 'message': 'Product updated successfully', + 'product': product + }) + + except Exception as e: + da.rollback() + return jsonify({ + 'status': 'error', + 'message': str(e) + }), 500 + +@products_bp.route('/api/product-attributes', methods=['GET']) +@login_required +def get_product_attributes(): + """Get product attributes configuration""" + attributes = da.get_product_attributes() + + return jsonify({ + 'status': 'success', + 'attributes': attributes + }) + @products_bp.route('/css/') def serve_css(filename): """Serve CSS files""" diff --git a/app/blueprints/users.py b/app/blueprints/users.py index 330e9d8..72974f8 100644 --- a/app/blueprints/users.py +++ b/app/blueprints/users.py @@ -3,33 +3,14 @@ User Management Blueprint Handles user CRUD operations """ from flask import Blueprint, render_template, request, jsonify, send_file, session -from werkzeug.security import generate_password_hash, check_password_hash +from werkzeug.security import check_password_hash from blueprints.auth import login_required, can_user, get_current_user +import data_access as da from io import BytesIO import json -import os users_bp = Blueprint('users', __name__, url_prefix='/users') -# Path to users file -USERS_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data', 'users.json') - -def load_users(): - """Load users from JSON file""" - if os.path.exists(USERS_FILE): - try: - with open(USERS_FILE, 'r') as f: - return json.load(f) - except: - return [] - return [] - -def save_users(users): - """Save users to JSON file""" - os.makedirs(os.path.dirname(USERS_FILE), exist_ok=True) - with open(USERS_FILE, 'w') as f: - json.dump(users, f, indent=2) - def permission_required(permission): """Decorator to require specific permission""" from functools import wraps @@ -52,15 +33,14 @@ def user_manager(): """Serve the user management page""" if not can_user('manage_users'): return "

403 - Permission Denied

You don't have permission to manage users.

", 403 - current_user_id = session.get('user_id', -1) - return render_template('user_manager.html', current_user_id=current_user_id) + return render_template('user_manager.html') @users_bp.route('/api', methods=['GET']) @login_required @permission_required('manage_users') def get_users(): """Get all users""" - users = load_users() + users = da.get_all_users() return jsonify({ 'status': 'success', 'users': users @@ -81,40 +61,16 @@ def add_user(): 'message': 'Username, password, and default location are required' }), 400 - users = load_users() + # Create user via data access layer + new_user = da.create_user(data) - # Check if username already exists - if any(user['username'] == data['username'] for user in users): + if not new_user: return jsonify({ 'status': 'error', 'message': 'Username already exists' }), 400 - # Hash the password - hashed_password = generate_password_hash(data['password'], method='pbkdf2:sha256') - - # Get location settings - location_settings = data.get('locationSettings', {}) - default_location = 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 new user - new_user = { - 'username': data['username'], - 'password': hashed_password, - 'defaultLocation': default_location, - 'locationSettings': location_settings, - 'permissions': data.get('permissions', {}), - 'active': data.get('active', True) - } - - users.append(new_user) - save_users(users) - + users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User added successfully', @@ -122,28 +78,25 @@ def add_user(): }) except Exception as e: + da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 -@users_bp.route('/api/', methods=['DELETE']) +@users_bp.route('/api/', methods=['DELETE']) @login_required @permission_required('manage_users') -def delete_user(user_index): - """Delete a specific user by index""" +def delete_user(user_id): + """Delete a specific user by ID""" try: - users = load_users() - - if user_index < 0 or user_index >= len(users): + if not da.delete_user(user_id): return jsonify({ 'status': 'error', - 'message': 'Invalid user index' - }), 400 - - users.pop(user_index) - save_users(users) + 'message': 'User not found' + }), 404 + users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User deleted successfully', @@ -151,29 +104,28 @@ def delete_user(user_index): }) except Exception as e: + da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 -@users_bp.route('/api//active', methods=['PATCH']) +@users_bp.route('/api//active', methods=['PATCH']) @login_required @permission_required('manage_users') -def toggle_user_active(user_index): +def toggle_user_active(user_id): """Toggle user active status""" try: data = request.json - users = load_users() + user = da.toggle_user_active(user_id, data.get('active', True)) - if user_index < 0 or user_index >= len(users): + if not user: return jsonify({ 'status': 'error', - 'message': 'Invalid user index' - }), 400 - - users[user_index]['active'] = data.get('active', True) - save_users(users) + 'message': 'User not found' + }), 404 + users = da.get_all_users() return jsonify({ 'status': 'success', 'message': 'User status updated', @@ -181,15 +133,16 @@ def toggle_user_active(user_index): }) except Exception as e: + da.rollback() return jsonify({ 'status': 'error', 'message': str(e) }), 500 -@users_bp.route('/api//change-password', methods=['POST']) +@users_bp.route('/api//change-password', methods=['POST']) @login_required @permission_required('manage_users') -def change_user_password(user_index): +def change_user_password(user_id): """Change a user's password with current user verification""" try: data = request.json @@ -222,26 +175,21 @@ def change_user_password(user_index): 'message': 'Your password is incorrect. Password change denied for security reasons.' }), 403 - # Load users and validate index - users = load_users() - if user_index < 0 or user_index >= len(users): + # Update password via data access layer + updated_user = da.change_user_password(user_id, new_password) + if not updated_user: return jsonify({ 'status': 'error', - 'message': 'Invalid user index' - }), 400 - - # Update password - hashed_password = generate_password_hash(new_password, method='pbkdf2:sha256') - target_username = users[user_index]['username'] - users[user_index]['password'] = hashed_password - save_users(users) + 'message': 'User not found' + }), 404 return jsonify({ 'status': 'success', - 'message': f'Password changed successfully for user "{target_username}"' + 'message': f'Password changed successfully for user "{updated_user["username"]}"' }) except Exception as e: + da.rollback() return jsonify({ 'status': 'error', 'message': str(e) @@ -253,7 +201,7 @@ def change_user_password(user_index): def download_users(): """Download users as JSON file""" try: - users = load_users() + users = da.get_all_users() if not users: return jsonify({ diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..eba5d2f --- /dev/null +++ b/app/config.py @@ -0,0 +1,24 @@ +""" +Application Configuration +Configure database backend and other settings +""" +import os + +# Database Configuration +# Set to True to use SQLite, False to use JSON files +USE_DATABASE = False + +# Database URI for SQLite +basedir = os.path.abspath(os.path.dirname(__file__)) +DATABASE_URI = f"sqlite:///{os.path.join(basedir, 'data', 'products.db')}" + +# JSON data directory +DATA_DIR = os.path.join(basedir, 'data') + +# Session configuration +SESSION_COOKIE_HTTPONLY = True +SESSION_COOKIE_SAMESITE = 'Lax' + +# SQLAlchemy configuration +SQLALCHEMY_TRACK_MODIFICATIONS = False +SQLALCHEMY_ECHO = False # Set to True for SQL debugging diff --git a/app/data_access.py b/app/data_access.py new file mode 100644 index 0000000..c1d8d30 --- /dev/null +++ b/app/data_access.py @@ -0,0 +1,91 @@ +""" +Data Access Layer +Automatically loads the correct data access module based on configuration +""" +import config + +# Import the appropriate data access module based on config +if config.USE_DATABASE: + print("Using SQLite database for data storage") + from data_access_sqlite import ( + # User operations + get_user_by_username, + get_user_by_id, + get_all_users, + create_user, + update_user, + delete_user, + toggle_user_active, + change_user_password, + # Product operations + get_all_products, + get_product_by_code, + create_product, + update_product, + delete_product, + # Location operations + get_all_locations, + get_location_by_code, + # Product attributes operations + get_product_attributes, + update_product_attributes, + # Utility functions + commit, + rollback + ) +else: + print("Using JSON files for data storage") + from data_access_json import ( + # User operations + get_user_by_username, + get_user_by_id, + get_all_users, + create_user, + update_user, + delete_user, + toggle_user_active, + change_user_password, + # Product operations + get_all_products, + get_product_by_code, + create_product, + update_product, + delete_product, + # Location operations + get_all_locations, + get_location_by_code, + # Product attributes operations + get_product_attributes, + update_product_attributes, + # Utility functions + commit, + rollback + ) + +# Re-export all functions so blueprints can do: from data_access import get_user_by_username +__all__ = [ + # User operations + 'get_user_by_username', + 'get_user_by_id', + 'get_all_users', + 'create_user', + 'update_user', + 'delete_user', + 'toggle_user_active', + 'change_user_password', + # Product operations + 'get_all_products', + 'get_product_by_code', + 'create_product', + 'update_product', + 'delete_product', + # Location operations + 'get_all_locations', + 'get_location_by_code', + # Product attributes operations + 'get_product_attributes', + 'update_product_attributes', + # Utility functions + 'commit', + 'rollback' +] diff --git a/app/data_access_json.py b/app/data_access_json.py new file mode 100644 index 0000000..f0fe741 --- /dev/null +++ b/app/data_access_json.py @@ -0,0 +1,449 @@ +""" +JSON Data Access Layer +Provides data operations using JSON files +""" +import json +import os +from werkzeug.security import generate_password_hash +from threading import Lock + +# File paths +basedir = os.path.abspath(os.path.dirname(__file__)) +USERS_FILE = os.path.join(basedir, 'data', 'users.json') +PRODUCTS_FILE = os.path.join(basedir, 'data', 'products.json') +PRODUCT_ATTRIBUTES_FILE = os.path.join(basedir, 'data', 'product_attributes.json') + +# Thread locks for file operations +users_lock = Lock() +products_lock = Lock() +attributes_lock = Lock() + +# ============================================================================ +# HELPER FUNCTIONS +# ============================================================================ + +def _load_json_file(filepath): + """Load a JSON file""" + if os.path.exists(filepath): + try: + with open(filepath, 'r', encoding='utf-8') as f: + return json.load(f) + except: + return [] + return [] + +def _save_json_file(filepath, data): + """Save data to a JSON file""" + os.makedirs(os.path.dirname(filepath), exist_ok=True) + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2) + +def _find_user_by_username(users, username): + """Find user index by username""" + for i, user in enumerate(users): + if user['username'] == username: + return i, user + return None, None + +def _find_product_by_code(products, product_code): + """Find product index by code""" + for i, product in enumerate(products): + if product['productCode'] == product_code: + return i, product + return None, None + +# ============================================================================ +# USER OPERATIONS +# ============================================================================ + +def get_user_by_username(username): + """Get a user by username""" + with users_lock: + users = _load_json_file(USERS_FILE) + _, user = _find_user_by_username(users, username) + return user + +def get_user_by_id(user_id): + """Get a user by ID (array index in JSON)""" + with users_lock: + users = _load_json_file(USERS_FILE) + if 0 <= user_id < len(users): + return users[user_id] + return None + +def get_all_users(): + """Get all users""" + with users_lock: + return _load_json_file(USERS_FILE) + +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 + """ + with users_lock: + users = _load_json_file(USERS_FILE) + + # Check if username already exists + if any(user['username'] == user_data['username'] for user in users): + 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 new user + new_user = { + 'username': user_data['username'], + 'password': password, + 'defaultLocation': default_location, + 'locationSettings': location_settings, + 'permissions': user_data.get('permissions', {}), + 'active': user_data.get('active', True) + } + + if user_data.get('superAdmin'): + new_user['superAdmin'] = True + + users.append(new_user) + _save_json_file(USERS_FILE, users) + + return new_user + +def update_user(user_id, user_data): + """ + Update an existing user + + Args: + user_id (int): User ID (array index) + user_data (dict): Updated user data + + Returns: + dict: Updated user as dictionary or None if not found + """ + with users_lock: + users = _load_json_file(USERS_FILE) + + if user_id < 0 or user_id >= len(users): + return None + + user = users[user_id] + + # Update fields + if 'defaultLocation' in user_data: + user['defaultLocation'] = user_data['defaultLocation'] + if 'active' in user_data: + user['active'] = user_data['active'] + if 'superAdmin' in user_data: + user['superAdmin'] = 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 + + _save_json_file(USERS_FILE, users) + return user + +def delete_user(user_id): + """ + Delete a user + + Args: + user_id (int): User ID (array index) + + Returns: + bool: True if deleted, False if not found + """ + with users_lock: + users = _load_json_file(USERS_FILE) + + if user_id < 0 or user_id >= len(users): + return False + + users.pop(user_id) + _save_json_file(USERS_FILE, users) + return True + +def toggle_user_active(user_id, active): + """ + Toggle user active status + + Args: + user_id (int): User ID (array index) + active (bool): Active status + + Returns: + dict: Updated user or None if not found + """ + with users_lock: + users = _load_json_file(USERS_FILE) + + if user_id < 0 or user_id >= len(users): + return None + + users[user_id]['active'] = active + _save_json_file(USERS_FILE, users) + return users[user_id] + +def change_user_password(user_id, new_password): + """ + Change a user's password + + Args: + user_id (int): User ID (array index) + new_password (str): New password (will be hashed) + + Returns: + dict: Updated user or None if not found + """ + with users_lock: + users = _load_json_file(USERS_FILE) + + if user_id < 0 or user_id >= len(users): + return None + + hashed_password = generate_password_hash(new_password, method='pbkdf2:sha256') + users[user_id]['password'] = hashed_password + _save_json_file(USERS_FILE, users) + return users[user_id] + +# ============================================================================ +# 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 + """ + with products_lock: + products = _load_json_file(PRODUCTS_FILE) + + if location: + products = [p for p in products if p.get('location') == location] + + return 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 + """ + with products_lock: + products = _load_json_file(PRODUCTS_FILE) + _, product = _find_product_by_code(products, product_code) + return product + +def create_product(product_data): + """ + Create a new product + + Args: + product_data (dict): Product data + + Returns: + dict: Created product or None if code exists + """ + with products_lock: + products = _load_json_file(PRODUCTS_FILE) + + # Check if product already exists + if any(p['productCode'] == product_data['productCode'] for p in products): + return None + + # Create product with normalized structure + new_product = { + 'id': product_data['productCode'], + 'productCode': product_data['productCode'], + 'category': product_data.get('category'), + 'description': product_data.get('description'), + 'discontinued': product_data.get('discontinued', False), + 'location': product_data.get('location'), + 'baseType': product_data.get('baseType', ''), + 'subType': product_data.get('subType', {}), + 'materials': product_data.get('materials', []), + 'colors': product_data.get('colors', []), + 'isAccessory': product_data.get('isAccessory', False), + 'compatibleAccessories': product_data.get('compatibleAccessories', []), + 'imageConfig': product_data.get('imageConfig', {}) + } + + products.append(new_product) + _save_json_file(PRODUCTS_FILE, products) + + return new_product + +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 + """ + with products_lock: + products = _load_json_file(PRODUCTS_FILE) + index, product = _find_product_by_code(products, product_code) + + if product is None: + 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['baseType'] = product_data['baseType'] + if 'isAccessory' in product_data: + product['isAccessory'] = product_data['isAccessory'] + 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'] + + _save_json_file(PRODUCTS_FILE, products) + return product + +def delete_product(product_code): + """ + Delete a product + + Args: + product_code (str): Product code + + Returns: + bool: True if deleted, False if not found + """ + with products_lock: + products = _load_json_file(PRODUCTS_FILE) + index, _ = _find_product_by_code(products, product_code) + + if index is None: + return False + + products.pop(index) + _save_json_file(PRODUCTS_FILE, products) + return True + +# ============================================================================ +# LOCATION OPERATIONS +# ============================================================================ + +def get_all_locations(): + """Get all locations""" + attributes = _load_json_file(PRODUCT_ATTRIBUTES_FILE) + return attributes.get('locations', []) + +def get_location_by_code(code): + """Get a location by code""" + locations = get_all_locations() + for loc in locations: + if loc['code'] == code: + return loc + return None + +# ============================================================================ +# PRODUCT ATTRIBUTES OPERATIONS +# ============================================================================ + +def get_product_attributes(): + """ + Get product attributes configuration + + Returns: + dict: Attributes with locations, statuses, productTypes, codeModifiers + """ + with attributes_lock: + attributes = _load_json_file(PRODUCT_ATTRIBUTES_FILE) + + if not attributes: + return { + 'locations': [], + 'statuses': [], + 'productTypes': [], + 'codeModifiers': [] + } + + return attributes + +def update_product_attributes(attributes_data): + """ + Update product attributes configuration + + Args: + attributes_data (dict): Updated attributes + + Returns: + dict: Updated attributes + """ + with attributes_lock: + attributes = _load_json_file(PRODUCT_ATTRIBUTES_FILE) + + 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'] + if 'locations' in attributes_data: + attributes['locations'] = attributes_data['locations'] + + _save_json_file(PRODUCT_ATTRIBUTES_FILE, attributes) + return attributes + +# ============================================================================ +# UTILITY FUNCTIONS +# ============================================================================ + +def commit(): + """No-op for JSON (writes are immediate)""" + pass + +def rollback(): + """No-op for JSON (no transactions)""" + pass diff --git a/app/data_access_sqlite.py b/app/data_access_sqlite.py new file mode 100644 index 0000000..d7dcf79 --- /dev/null +++ b/app/data_access_sqlite.py @@ -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() diff --git a/app/migrate_to_sqlite.py b/app/migrate_to_sqlite.py new file mode 100644 index 0000000..ce90911 --- /dev/null +++ b/app/migrate_to_sqlite.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 +""" +Migration script to convert JSON data to SQLite database. +Run this script to migrate data from JSON files to the new SQLite database. +""" + +import json +import os +import sys +from datetime import datetime + +# Add the app directory to the path +basedir = os.path.abspath(os.path.dirname(__file__)) +sys.path.insert(0, basedir) + +from app import app +from models import db, User, Product, Location, ProductAttribute, Question, Accessory + +def load_json_file(filename): + """Load a JSON file from the data directory""" + filepath = os.path.join(basedir, 'data', filename) + if not os.path.exists(filepath): + print(f"Warning: {filename} not found, skipping...") + return None + + try: + with open(filepath, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + print(f"Error loading {filename}: {e}") + return None + +def migrate_users(): + """Migrate users from JSON to database""" + print("Migrating users...") + users_data = load_json_file('users.json') + if not users_data: + return + + count = 0 + for user_data in users_data: + # Check if user already exists + existing = User.query.filter_by(username=user_data['username']).first() + if existing: + print(f" User '{user_data['username']}' already exists, skipping...") + continue + + user = User( + username=user_data['username'], + password=user_data['password'], + default_location=user_data.get('defaultLocation'), + active=user_data.get('active', True), + super_admin=user_data.get('superAdmin', False) + ) + user.permissions = user_data.get('permissions', {}) + user.locationSettings = user_data.get('locationSettings', {}) + + db.session.add(user) + count += 1 + + db.session.commit() + print(f" ✓ Migrated {count} users") + +def migrate_locations(): + """Migrate locations from product_attributes.json""" + print("Migrating locations...") + attributes_data = load_json_file('product_attributes.json') + if not attributes_data or 'locations' not in attributes_data: + return + + count = 0 + for loc_data in attributes_data['locations']: + # Check if location already exists + existing = Location.query.filter_by(code=loc_data['code']).first() + if existing: + print(f" Location '{loc_data['code']}' already exists, skipping...") + continue + + location = Location( + code=loc_data['code'], + name=loc_data['name'], + active=True + ) + db.session.add(location) + count += 1 + + db.session.commit() + print(f" ✓ Migrated {count} locations") + +def migrate_product_attributes(): + """Migrate product attributes from JSON""" + print("Migrating product attributes...") + attributes_data = load_json_file('product_attributes.json') + if not attributes_data: + return + + # Check if attributes already exist + existing = ProductAttribute.query.first() + if existing: + print(" Product attributes already exist, updating...") + attr = existing + else: + attr = ProductAttribute() + db.session.add(attr) + + # Set the attributes + attr.statuses = attributes_data.get('statuses', []) + attr.productTypes = attributes_data.get('productTypes', []) + attr.codeModifiers = attributes_data.get('codeModifiers', []) + + db.session.commit() + print(f" ✓ Product attributes migrated") + +def migrate_products(): + """Migrate products from JSON to database""" + print("Migrating products...") + products_data = load_json_file('products.json') + if not products_data: + return + + count = 0 + skipped = 0 + for prod_data in products_data: + # Check if product already exists + existing = Product.query.filter_by(product_code=prod_data['productCode']).first() + if existing: + skipped += 1 + continue + + product = Product( + product_code=prod_data['productCode'], + category=prod_data.get('category'), + description=prod_data.get('description'), + discontinued=prod_data.get('discontinued', False), + location=prod_data.get('location'), + base_type=prod_data.get('baseType', ''), + is_accessory=prod_data.get('isAccessory', False) + ) + + # Set JSON properties + product.subType = prod_data.get('subType', {}) + product.materials = prod_data.get('materials', []) + product.colors = prod_data.get('colors', []) + product.compatibleAccessories = prod_data.get('compatibleAccessories', []) + product.imageConfig = prod_data.get('imageConfig', {}) + + db.session.add(product) + count += 1 + + db.session.commit() + print(f" ✓ Migrated {count} products (skipped {skipped} existing)") + +def migrate_questions(): + """Migrate questions from JSON to database""" + print("Migrating questions...") + questions_data = load_json_file('questions.json') + if not questions_data: + return + + # Check if questions already exist + existing_count = Question.query.count() + if existing_count > 0: + print(f" Questions already exist ({existing_count} records), skipping...") + return + + count = 0 + for idx, question_data in enumerate(questions_data): + question = Question( + question_data_json=json.dumps(question_data), + order=idx, + active=True + ) + db.session.add(question) + count += 1 + + db.session.commit() + print(f" ✓ Migrated {count} questions") + +def migrate_accessories(): + """Migrate accessories from JSON to database""" + print("Migrating accessories...") + accessories_data = load_json_file('accessories.json') + if not accessories_data: + return + + # Check if accessories already exist + existing_count = Accessory.query.count() + if existing_count > 0: + print(f" Accessories already exist ({existing_count} records), skipping...") + return + + count = 0 + for category, items in accessories_data.items(): + for item_data in items: + accessory = Accessory( + accessory_data_json=json.dumps(item_data), + category=category, + active=True + ) + db.session.add(accessory) + count += 1 + + db.session.commit() + print(f" ✓ Migrated {count} accessories") + +def backup_json_files(): + """Create backup of JSON files before migration""" + print("Creating backup of JSON files...") + backup_dir = os.path.join(basedir, 'data', 'json_backup') + os.makedirs(backup_dir, exist_ok=True) + + json_files = ['users.json', 'products.json', 'product_attributes.json', + 'questions.json', 'accessories.json', 'navigation.json', + 'filter_config.json', 'product_bitwise.json'] + + count = 0 + for filename in json_files: + src = os.path.join(basedir, 'data', filename) + if os.path.exists(src): + dst = os.path.join(backup_dir, f"{filename}.backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}") + import shutil + shutil.copy2(src, dst) + count += 1 + + print(f" ✓ Backed up {count} files to {backup_dir}") + +def main(): + """Main migration function""" + print("=" * 60) + print("JSON to SQLite Migration Script") + print("=" * 60) + print() + + # Create application context + with app.app_context(): + # Create all tables + print("Creating database tables...") + db.create_all() + print(" ✓ Tables created") + print() + + # Create backup + backup_json_files() + print() + + # Run migrations + try: + migrate_users() + migrate_locations() + migrate_product_attributes() + migrate_products() + migrate_questions() + migrate_accessories() + + print() + print("=" * 60) + print("✅ Migration completed successfully!") + print("=" * 60) + print() + print("Database location:", os.path.join(basedir, 'data', 'products.db')) + print() + print("Next steps:") + print("1. Test the application to ensure everything works") + print("2. Update blueprints to use database instead of JSON") + print("3. Keep JSON files as backup or remove them") + + except Exception as e: + print() + print("=" * 60) + print("❌ Migration failed!") + print("=" * 60) + print(f"Error: {e}") + import traceback + traceback.print_exc() + db.session.rollback() + +if __name__ == '__main__': + main() diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..bb7ae0d --- /dev/null +++ b/app/models.py @@ -0,0 +1,262 @@ +from flask_sqlalchemy import SQLAlchemy +from datetime import datetime +import json + +db = SQLAlchemy() + +class User(db.Model): + __tablename__ = 'users' + + id = db.Column(db.Integer, primary_key=True) + username = db.Column(db.String(80), unique=True, nullable=False, index=True) + password = db.Column(db.String(255), nullable=False) + default_location = db.Column(db.String(10)) + active = db.Column(db.Boolean, default=True) + super_admin = db.Column(db.Boolean, default=False) + + # Permissions stored as JSON + permissions_json = db.Column(db.Text, default='{}') + location_settings_json = db.Column(db.Text, default='{}') + + created_at = db.Column(db.DateTime, default=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + @property + def permissions(self): + """Get permissions as dictionary""" + return json.loads(self.permissions_json) if self.permissions_json else {} + + @permissions.setter + def permissions(self, value): + """Set permissions from dictionary""" + self.permissions_json = json.dumps(value) + + @property + def locationSettings(self): + """Get location settings as dictionary""" + return json.loads(self.location_settings_json) if self.location_settings_json else {} + + @locationSettings.setter + def locationSettings(self, value): + """Set location settings from dictionary""" + self.location_settings_json = json.dumps(value) + + def to_dict(self): + """Convert user to dictionary matching JSON format""" + return { + 'username': self.username, + 'password': self.password, + 'defaultLocation': self.default_location, + 'permissions': self.permissions, + 'locationSettings': self.locationSettings, + 'active': self.active, + 'superAdmin': self.super_admin + } + + def __repr__(self): + return f'' + + +class Product(db.Model): + __tablename__ = 'products' + + id = db.Column(db.Integer, primary_key=True) + product_code = db.Column(db.String(50), unique=True, nullable=False, index=True) + category = db.Column(db.String(50)) + description = db.Column(db.Text) + discontinued = db.Column(db.Boolean, default=False) + location = db.Column(db.String(50)) + base_type = db.Column(db.String(50)) + + # Store complex fields as JSON + sub_type_json = db.Column(db.Text, default='{}') + materials_json = db.Column(db.Text, default='[]') + colors_json = db.Column(db.Text, default='[]') + compatible_accessories_json = db.Column(db.Text, default='[]') + image_config_json = db.Column(db.Text, default='{}') + + is_accessory = db.Column(db.Boolean, default=False) + + created_at = db.Column(db.DateTime, default=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + @property + def subType(self): + return json.loads(self.sub_type_json) if self.sub_type_json else {} + + @subType.setter + def subType(self, value): + self.sub_type_json = json.dumps(value) + + @property + def materials(self): + return json.loads(self.materials_json) if self.materials_json else [] + + @materials.setter + def materials(self, value): + self.materials_json = json.dumps(value) + + @property + def colors(self): + return json.loads(self.colors_json) if self.colors_json else [] + + @colors.setter + def colors(self, value): + self.colors_json = json.dumps(value) + + @property + def compatibleAccessories(self): + return json.loads(self.compatible_accessories_json) if self.compatible_accessories_json else [] + + @compatibleAccessories.setter + def compatibleAccessories(self, value): + self.compatible_accessories_json = json.dumps(value) + + @property + def imageConfig(self): + return json.loads(self.image_config_json) if self.image_config_json else {} + + @imageConfig.setter + def imageConfig(self, value): + self.image_config_json = json.dumps(value) + + def to_dict(self): + """Convert product to dictionary matching JSON format""" + return { + 'id': self.product_code, + 'productCode': self.product_code, + 'category': self.category, + 'description': self.description, + 'discontinued': self.discontinued, + 'location': self.location, + 'baseType': self.base_type, + 'subType': self.subType, + 'materials': self.materials, + 'colors': self.colors, + 'isAccessory': self.is_accessory, + 'compatibleAccessories': self.compatibleAccessories, + 'imageConfig': self.imageConfig + } + + def __repr__(self): + return f'' + + +class Location(db.Model): + __tablename__ = 'locations' + + id = db.Column(db.Integer, primary_key=True) + code = db.Column(db.String(10), unique=True, nullable=False, index=True) + name = db.Column(db.String(100), nullable=False) + active = db.Column(db.Boolean, default=True) + + created_at = db.Column(db.DateTime, default=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + def to_dict(self): + """Convert location to dictionary matching JSON format""" + return { + 'code': self.code, + 'name': self.name + } + + def __repr__(self): + return f'' + + +class ProductAttribute(db.Model): + __tablename__ = 'product_attributes' + + id = db.Column(db.Integer, primary_key=True) + + # Store all attribute data as JSON + statuses_json = db.Column(db.Text, default='[]') + product_types_json = db.Column(db.Text, default='[]') + code_modifiers_json = db.Column(db.Text, default='[]') + + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + @property + def statuses(self): + return json.loads(self.statuses_json) if self.statuses_json else [] + + @statuses.setter + def statuses(self, value): + self.statuses_json = json.dumps(value) + + @property + def productTypes(self): + return json.loads(self.product_types_json) if self.product_types_json else [] + + @productTypes.setter + def productTypes(self, value): + self.product_types_json = json.dumps(value) + + @property + def codeModifiers(self): + return json.loads(self.code_modifiers_json) if self.code_modifiers_json else [] + + @codeModifiers.setter + def codeModifiers(self, value): + self.code_modifiers_json = json.dumps(value) + + def to_dict(self): + """Convert attributes to dictionary matching JSON format""" + locations = Location.query.filter_by(active=True).all() + return { + 'locations': [loc.to_dict() for loc in locations], + 'statuses': self.statuses, + 'productTypes': self.productTypes, + 'codeModifiers': self.codeModifiers + } + + def __repr__(self): + return f'' + + +# Additional models for other JSON files if needed + +class Question(db.Model): + __tablename__ = 'questions' + + id = db.Column(db.Integer, primary_key=True) + question_data_json = db.Column(db.Text, nullable=False) + order = db.Column(db.Integer, default=0) + active = db.Column(db.Boolean, default=True) + + created_at = db.Column(db.DateTime, default=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + @property + def questionData(self): + return json.loads(self.question_data_json) if self.question_data_json else {} + + @questionData.setter + def questionData(self, value): + self.question_data_json = json.dumps(value) + + def __repr__(self): + return f'' + + +class Accessory(db.Model): + __tablename__ = 'accessories' + + id = db.Column(db.Integer, primary_key=True) + accessory_data_json = db.Column(db.Text, nullable=False) + category = db.Column(db.String(50)) + active = db.Column(db.Boolean, default=True) + + created_at = db.Column(db.DateTime, default=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + @property + def accessoryData(self): + return json.loads(self.accessory_data_json) if self.accessory_data_json else {} + + @accessoryData.setter + def accessoryData(self, value): + self.accessory_data_json = json.dumps(value) + + def __repr__(self): + return f'' diff --git a/app/requirements.txt b/app/requirements.txt index 02ae215..686ab81 100644 --- a/app/requirements.txt +++ b/app/requirements.txt @@ -1 +1,2 @@ Flask==3.1.3 +Flask-SQLAlchemy==3.0.5