import os import sys import time import sqlite3 from pathlib import Path from datetime import datetime, timedelta from sqlalchemy import inspect, text from sqlalchemy.exc import OperationalError from app.db.base import Base # Import all models from app.db.session import engine, db_file from app.core.config import settings def init_db() -> None: """ Initialize database with required tables directly using both SQLAlchemy and SQLite native commands for maximum reliability. """ print(f"Initializing database at {db_file}") db_dir = db_file.parent # Ensure database directory exists try: db_dir.mkdir(parents=True, exist_ok=True) print(f"Database directory created or already exists: {db_dir}") except Exception as e: print(f"Error creating database directory: {e}") # First, try to create an empty database file if it doesn't exist if not db_file.exists(): try: # Create an empty file db_file.touch() print(f"Created empty database file: {db_file}") except Exception as e: print(f"Failed to create database file: {e}") # First, try direct SQLite connection to create basic structure # This bypasses SQLAlchemy entirely for the initial database creation try: print(f"Attempting direct SQLite connection to {db_file}") sqlite_conn = sqlite3.connect(str(db_file)) sqlite_conn.execute("PRAGMA journal_mode=WAL") sqlite_conn.execute("CREATE TABLE IF NOT EXISTS _db_init_check (id INTEGER PRIMARY KEY)") sqlite_conn.execute("INSERT OR IGNORE INTO _db_init_check VALUES (1)") sqlite_conn.commit() sqlite_conn.close() print("Direct SQLite connection and initialization successful") except Exception as e: print(f"Direct SQLite initialization error: {e}") # Now try with SQLAlchemy try: # Try to connect to check if the database is accessible max_retries = 5 retry_count = 0 connected = False while retry_count < max_retries and not connected: try: with engine.connect() as conn: result = conn.execute(text("SELECT 1")).scalar() print(f"Database connection successful. Test query result: {result}") connected = True except Exception as e: retry_count += 1 print(f"Database connection error (attempt {retry_count}/{max_retries}): {e}") time.sleep(1) # Wait a second before retrying if not connected: print(f"Failed to connect to database after {max_retries} attempts") # Continue anyway to see if we can make progress # Try to create tables try: print("Creating database tables with SQLAlchemy...") Base.metadata.create_all(bind=engine) # Verify tables inspector = inspect(engine) tables = inspector.get_table_names() print(f"Tables in database: {', '.join(tables)}") if 'task' not in tables: print("WARNING: 'task' table not created!") else: print("'task' table successfully created.") except Exception as e: print(f"Error creating tables: {e}") # Print database info for debugging try: print(f"Database file size: {os.path.getsize(db_file)} bytes") print(f"Database file permissions: {oct(os.stat(db_file).st_mode)[-3:]}") print(f"Database dir permissions: {oct(os.stat(db_dir).st_mode)[-3:]}") except Exception as e: print(f"Error getting file info: {e}") print("Database initialization completed") except Exception as e: print(f"Database initialization error: {str(e)}") print("Continuing anyway...") def create_test_task(): """Create a test task in the database to verify everything is working.""" print("Attempting to create a test task...") try: # Try direct SQLite approach first try: conn = sqlite3.connect(str(db_file)) cursor = conn.cursor() # Check if task table exists cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task'") if not cursor.fetchone(): print("Task table doesn't exist - cannot create test task") return # Check if any tasks exist cursor.execute("SELECT COUNT(*) FROM task") count = cursor.fetchone()[0] if count == 0: # Create a task directly with SQLite now = datetime.utcnow().isoformat() cursor.execute( """ INSERT INTO task ( title, description, priority, status, completed, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( "Test Task (Direct SQL)", "This is a test task created directly with SQLite", "medium", "todo", 0, # not completed now, now ) ) conn.commit() task_id = cursor.lastrowid print(f"Created test task with direct SQLite, ID: {task_id}") else: print(f"Found {count} existing tasks, no need to create test task") conn.close() except Exception as e: print(f"Error with direct SQLite test task creation: {e}") # Continue with SQLAlchemy approach # Now try with SQLAlchemy try: from app.crud.task import task as task_crud from app.schemas.task import TaskCreate from app.db.session import SessionLocal db = SessionLocal() try: # Check if there are any tasks try: existing_tasks = db.execute(text("SELECT COUNT(*) FROM task")).scalar() if existing_tasks > 0: print(f"Test task not needed, found {existing_tasks} existing tasks") return except Exception as e: print(f"Error checking for existing tasks: {e}") # Continue anyway to try creating a task # Create a test task test_task = TaskCreate( title="Test Task (SQLAlchemy)", description="This is a test task created with SQLAlchemy", priority="medium", status="todo", due_date=datetime.utcnow() + timedelta(days=7), completed=False ) created_task = task_crud.create(db, obj_in=test_task) print(f"Created test task with SQLAlchemy, ID: {created_task.id}") finally: db.close() except Exception as e: print(f"Error with SQLAlchemy test task creation: {e}") except Exception as e: print(f"Global error creating test task: {e}") import traceback print(traceback.format_exc()) if __name__ == "__main__": init_db() create_test_task()