LocalFoodAI_FerRo988/database.py

import sqlite3
import os
import logging
from typing import Optional, Dict, Any

logger = logging.getLogger(__name__)

# Locate db correctly in the same directory
DB_PATH = os.path.join(os.path.dirname(__file__), "localfood.db")

def get_db_connection():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

def create_tables():
    """Initialize the SQLite database with required tables"""
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        
        # Create users table securely locally
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password_hash TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        conn.commit()
        conn.close()
        logger.info("Database and tables initialized successfully.")
    except Exception as e:
        logger.error(f"Error initializing database: {e}")
        raise

def get_user_by_username(username: str) -> Optional[Dict[str, Any]]:
    """Retrieve user dictionary if they exist"""
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE username = ?", (username,))
        row = cursor.fetchone()
        conn.close()
        return dict(row) if row else None
    except Exception as e:
        logger.error(f"Database error fetching user: {e}")
        return None

def create_user(username: str, password_hash: str) -> bool:
    """Creates a user securely. Returns True if successful, False if username exists."""
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO users (username, password_hash) VALUES (?, ?)",
            (username, password_hash)
        )
        conn.commit()
        conn.close()
        return True
    except sqlite3.IntegrityError:
        # Prevent duplicate usernames explicitly
        return False
    except Exception as e:
        logger.error(f"Database error during user creation: {e}")
        raise