| import sqlite3 |
| import os |
| from pathlib import Path |
|
|
| DB_PATH = os.path.join(os.path.dirname(__file__), "fitplan.db") |
|
|
| def init_database(): |
| """Initialize the SQLite database with users table.""" |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| |
| cursor.execute(''' |
| CREATE TABLE IF NOT EXISTS users ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| name TEXT NOT NULL, |
| email TEXT UNIQUE NOT NULL, |
| password_hash TEXT NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ) |
| ''') |
| |
| conn.commit() |
| conn.close() |
|
|
| def user_exists(email): |
| """Check if a user with the given email already exists.""" |
| email = email.strip().lower() |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| cursor.execute('SELECT id FROM users WHERE email = ?', (email,)) |
| result = cursor.fetchone() |
| conn.close() |
| return result is not None |
|
|
| def register_user(name, email, password_hash): |
| """Register a new user in the database.""" |
| email = email.strip().lower() |
| try: |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| cursor.execute(''' |
| INSERT INTO users (name, email, password_hash) |
| VALUES (?, ?, ?) |
| ''', (name, email, password_hash)) |
| conn.commit() |
| conn.close() |
| return True |
| except sqlite3.IntegrityError as e: |
| print(f"Database error: {e}") |
| return False |
|
|
| def get_user_by_email(email): |
| """Retrieve user data by email.""" |
| email = email.strip().lower() |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| cursor.execute('SELECT id, name, email, password_hash FROM users WHERE email = ?', (email,)) |
| user = cursor.fetchone() |
| conn.close() |
| |
| if user: |
| return { |
| "id": user[0], |
| "name": user[1], |
| "email": user[2], |
| "password_hash": user[3] |
| } |
| return None |
|
|
| def get_user_by_name_email(name, email): |
| """Retrieve user data by name and email.""" |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| cursor.execute('SELECT id, name, email, password_hash FROM users WHERE name = ? AND email = ?', (name, email)) |
| user = cursor.fetchone() |
| conn.close() |
| |
| if user: |
| return { |
| "id": user[0], |
| "name": user[1], |
| "email": user[2], |
| "password_hash": user[3] |
| } |
| return None |
|
|
| def get_all_users(): |
| """Debug: Get all users in database.""" |
| conn = sqlite3.connect(DB_PATH) |
| cursor = conn.cursor() |
| cursor.execute('SELECT id, name, email FROM users') |
| users = cursor.fetchall() |
| conn.close() |
| return users |
| user = cursor.fetchone() |
| conn.close() |
| |
| if user: |
| return { |
| "id": user[0], |
| "name": user[1], |
| "email": user[2], |
| "password_hash": user[3] |
| } |
| return None |
