backend.im/taskmanagerapi-ttkjqk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement ultra-reliable database access with comprehensive error handling

Browse Source
This commit is contained in:
Automated Action 2025-05-16 06:46:45 +00:00
parent 669d16ace5
commit 0645953951
5 changed files with 512 additions and 312 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

272
app/api/routers/tasks.py
View File

@ -94,136 +94,204 @@ def create_task(
task_in: TaskCreate, task_in: TaskCreate,
) -> Any: ) -> Any:
""" """
Create new task. Create new task - using direct SQLite approach for reliability.
""" """
import sqlite3 import sqlite3
import json import time
import sys
import traceback import traceback
from datetime import datetime from datetime import datetime
from app.db.session import db_file from app.db.session import db_file
# Print detailed request info for debugging # Log creation attempt
print(f"[{datetime.now().isoformat()}] Creating task: {task_in}") print(f"[{datetime.now().isoformat()}] Task creation requested", file=sys.stdout)
# Use direct SQLite for maximum reliability
try: try:
# Handle datetime conversion for due_date # Extract task data regardless of Pydantic version
if task_in.due_date:
print(f"Due date before processing: {task_in.due_date}")
if isinstance(task_in.due_date, str):
try:
task_in.due_date = datetime.fromisoformat(task_in.due_date.replace('Z', '+00:00'))
except Exception as e:
print(f"Error parsing due_date: {e}")
# Continue with string, SQLAlchemy will handle it
# First try the normal SQLAlchemy approach
try: try:
print("Attempting to create task via SQLAlchemy...") if hasattr(task_in, 'model_dump'):
task = crud.task.create(db, obj_in=task_in) task_data = task_in.model_dump()
print(f"Task created successfully with ID: {task.id}") elif hasattr(task_in, 'dict'):
return task task_data = task_in.dict()
except Exception as e: else:
print(f"SQLAlchemy task creation failed: {e}") # Fallback for any case
print(traceback.format_exc()) task_data = {
# Continue to try direct SQLite approach 'title': getattr(task_in, 'title', 'Untitled Task'),
'description': getattr(task_in, 'description', ''),
# Fallback: Try direct SQLite approach 'priority': getattr(task_in, 'priority', 'medium'),
print("Falling back to direct SQLite task creation...") 'status': getattr(task_in, 'status', 'todo'),
try: 'due_date': getattr(task_in, 'due_date', None),
# Convert Pydantic model to dict 'completed': getattr(task_in, 'completed', False)
task_data = task_in.model_dump() if hasattr(task_in, 'model_dump') else task_in.dict() }
print(f"Task data: {task_data}") print(f"Task data: {task_data}")
except Exception as e:
# Format datetime objects print(f"Error extracting task data: {e}")
if task_data.get('due_date'): # Fallback to minimal data
task_data = {
'title': str(getattr(task_in, 'title', 'Unknown Title')),
'description': str(getattr(task_in, 'description', '')),
'priority': 'medium',
'status': 'todo',
'completed': False
}
# Format due_date if present
if task_data.get('due_date'):
try:
if isinstance(task_data['due_date'], datetime): if isinstance(task_data['due_date'], datetime):
task_data['due_date'] = task_data['due_date'].isoformat() task_data['due_date'] = task_data['due_date'].isoformat()
elif isinstance(task_data['due_date'], str):
# Connect directly to SQLite and insert the task # Standardize format by parsing and reformatting
conn = sqlite3.connect(str(db_file)) parsed_date = datetime.fromisoformat(
cursor = conn.cursor() task_data['due_date'].replace('Z', '+00:00')
now = datetime.utcnow().isoformat() )
task_data['due_date'] = parsed_date.isoformat()
# Check if task table exists except Exception as e:
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task'") print(f"Warning: Could not parse due_date: {e}")
if not cursor.fetchone(): # Keep as-is or set to None if invalid
# Create the task table if it doesn't exist if not isinstance(task_data['due_date'], str):
task_data['due_date'] = None
# Get current timestamp for created/updated fields
now = datetime.utcnow().isoformat()
# Connect to SQLite with retry logic
for retry in range(3):
conn = None
try:
# Try to connect to the database with a timeout
conn = sqlite3.connect(str(db_file), timeout=30)
cursor = conn.cursor()
# Create the task table if it doesn't exist - using minimal schema
cursor.execute(""" cursor.execute("""
CREATE TABLE IF NOT EXISTS task ( CREATE TABLE IF NOT EXISTS task (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL, title TEXT NOT NULL,
description TEXT, description TEXT,
priority TEXT, priority TEXT DEFAULT 'medium',
status TEXT, status TEXT DEFAULT 'todo',
due_date TEXT, due_date TEXT,
completed INTEGER, completed INTEGER DEFAULT 0,
created_at TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT updated_at TEXT DEFAULT CURRENT_TIMESTAMP
) )
""") """)
print("Created task table with direct SQLite")
# Insert the task - provide defaults for all fields
# Insert the task cursor.execute(
cursor.execute( """
""" INSERT INTO task (
INSERT INTO task ( title, description, priority, status,
title, description, priority, status, due_date, completed, created_at, updated_at
due_date, completed, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """,
""", (
( task_data.get('title', 'Untitled'),
task_data.get('title', 'Untitled Task'), task_data.get('description', ''),
task_data.get('description', ''), task_data.get('priority', 'medium'),
task_data.get('priority', 'medium'), task_data.get('status', 'todo'),
task_data.get('status', 'todo'), task_data.get('due_date'),
task_data.get('due_date'), 1 if task_data.get('completed') else 0,
1 if task_data.get('completed') else 0, now,
now, now
now )
) )
)
conn.commit()
task_id = cursor.lastrowid
# Return the created task
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
row = cursor.fetchone()
conn.close()
if row:
column_names = ['id', 'title', 'description', 'priority', 'status',
'due_date', 'completed', 'created_at', 'updated_at']
task_dict = {column_names[i]: row[i] for i in range(len(column_names))}
# Convert completed to boolean
task_dict['completed'] = bool(task_dict['completed'])
# Fake Task object with attributes # Get the ID of the inserted task
class TaskResult: task_id = cursor.lastrowid
def __init__(self, **kwargs): print(f"Task inserted with ID: {task_id}")
for key, value in kwargs.items():
setattr(self, key, value)
print(f"Created task with direct SQLite, ID: {task_id}") # Commit the transaction
return TaskResult(**task_dict) conn.commit()
else:
raise Exception("Task was created but could not be retrieved") # Retrieve the created task to return it
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
row = cursor.fetchone()
if row:
# Get column names from cursor description
column_names = [desc[0] for desc in cursor.description]
# Create a dictionary from row values
task_dict = dict(zip(column_names, row))
# Convert 'completed' to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
# Create an object that mimics the Task model
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
print(f"Task created successfully: ID={task_id}")
# Close the connection and return the task
conn.close()
return TaskResult(**task_dict)
else:
conn.close()
raise Exception(f"Task creation succeeded but retrieval failed for ID: {task_id}")
except sqlite3.OperationalError as e:
if conn:
conn.close()
# Check if retry is appropriate
if "database is locked" in str(e) and retry < 2:
wait_time = (retry + 1) * 1.5 # Exponential backoff
print(f"Database locked, retrying in {wait_time}s (attempt {retry+1}/3)")
time.sleep(wait_time)
else:
print(f"SQLite operational error: {e}")
raise
except Exception as e:
if conn:
# Try to rollback if connection is still open
try:
conn.rollback()
except:
pass
conn.close()
print(f"Error in SQLite task creation: {e}")
print(traceback.format_exc())
# Only retry on specific transient errors
if retry < 2 and ("locked" in str(e).lower() or "busy" in str(e).lower()):
time.sleep(1)
continue
raise
# If we reach here, the retry loop failed
raise Exception("Failed to create task after multiple attempts")
except Exception as sqlite_error:
# Final fallback: try SQLAlchemy approach
try:
print(f"Direct SQLite approach failed: {sqlite_error}")
print("Trying SQLAlchemy as fallback...")
except Exception as e: task = crud.task.create(db, obj_in=task_in)
print(f"Direct SQLite task creation failed: {e}") print(f"Task created with SQLAlchemy fallback: ID={task.id}")
return task
except Exception as alch_error:
print(f"SQLAlchemy fallback also failed: {alch_error}")
print(traceback.format_exc()) print(traceback.format_exc())
# Provide detailed error information
error_detail = f"Task creation failed. Primary error: {str(sqlite_error)}. Fallback error: {str(alch_error)}"
print(f"Final error: {error_detail}")
raise HTTPException( raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"All task creation methods failed. Last error: {str(e)}", detail=error_detail
) )
except HTTPException:
raise
except Exception as e:
print(f"Unhandled error in create_task: {str(e)}")
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error creating task: {str(e)}",
)
@router.get("/{task_id}", response_model=Task) @router.get("/{task_id}", response_model=Task)

45
app/core/config.py
View File

@ -5,33 +5,28 @@ from typing import Any, Dict, List, Optional, Union
from pydantic import AnyHttpUrl, field_validator from pydantic import AnyHttpUrl, field_validator
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
# Try multiple possible database locations in order of preference import os
DB_LOCATIONS = [
Path("/app/storage/db"),
Path("/tmp/taskmanager/db"),
Path.home() / ".taskmanager/db",
Path.cwd() / "db"
]
# Try to find or create a writable database directory # Use a simple, universally accessible path for database
DB_DIR = None # First try environment variable, then local directory
for location in DB_LOCATIONS: DB_PATH = os.environ.get("DB_PATH")
try: if DB_PATH:
location.mkdir(parents=True, exist_ok=True) DB_DIR = Path(DB_PATH)
# Check if directory is writable by creating a test file print(f"Using database path from environment: {DB_DIR}")
test_file = location / "test_write" else:
test_file.touch() # Use 'db' directory in the current working directory
test_file.unlink() # Remove test file DB_DIR = Path.cwd() / "db"
DB_DIR = location print(f"Using local database path: {DB_DIR}")
print(f"Using database directory: {DB_DIR}")
break
except Exception as e:
print(f"Could not use {location} for database: {e}")
# If no location works, fall back to current directory # Create database directory
if DB_DIR is None: try:
DB_DIR = Path.cwd() DB_DIR.mkdir(parents=True, exist_ok=True)
print(f"Falling back to current directory for database: {DB_DIR}") print(f"Created or verified database directory: {DB_DIR}")
except Exception as e:
print(f"Error creating database directory: {e}")
# Fall back to /tmp if we can't create our preferred directory
DB_DIR = Path("/tmp")
print(f"Falling back to temporary directory: {DB_DIR}")
class Settings(BaseSettings): class Settings(BaseSettings):
PROJECT_NAME: str = "Task Manager API" PROJECT_NAME: str = "Task Manager API"

183
app/db/init_db.py
View File

@ -15,95 +15,126 @@ from app.core.config import settings
def init_db() -> None: def init_db() -> None:
""" """
Initialize database with required tables directly using both SQLAlchemy and Initialize database using both direct SQLite and SQLAlchemy approaches
SQLite native commands for maximum reliability. for maximum reliability.
""" """
print(f"Initializing database at {db_file}") print(f"Initializing database at {db_file}")
db_dir = db_file.parent print(f"Using SQLAlchemy URL: {settings.SQLALCHEMY_DATABASE_URL}")
# Ensure database directory exists # First try direct SQLite approach to ensure we have a basic database file
try: try:
db_dir.mkdir(parents=True, exist_ok=True) # Ensure database file exists and is writable
print(f"Database directory created or already exists: {db_dir}") with open(db_file, 'a'): # Try opening for append (creates if doesn't exist)
except Exception as e: os.utime(db_file, None) # Update access/modify time
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: print(f"Database file exists and is writable: {db_file}")
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: # Try direct SQLite connection to create task table
print(f"Failed to connect to database after {max_retries} attempts") conn = sqlite3.connect(str(db_file))
# Continue anyway to see if we can make progress
# Try to create tables # Enable foreign keys and WAL journal mode
try: conn.execute("PRAGMA foreign_keys = ON")
print("Creating database tables with SQLAlchemy...") conn.execute("PRAGMA journal_mode = WAL")
Base.metadata.create_all(bind=engine)
# Create task table if it doesn't exist
conn.execute("""
CREATE TABLE IF NOT EXISTS task (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT,
priority TEXT DEFAULT 'medium',
status TEXT DEFAULT 'todo',
due_date TEXT,
completed INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Create an index on the id column
conn.execute("CREATE INDEX IF NOT EXISTS idx_task_id ON task(id)")
# Add a sample task if the table is empty
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM task")
count = cursor.fetchone()[0]
if count == 0:
now = datetime.utcnow().isoformat()
conn.execute("""
INSERT INTO task (title, description, priority, status, completed, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
"Example Task",
"This is an example task created during initialization",
"medium",
"todo",
0,
now,
now
))
# Verify tables conn.commit()
inspector = inspect(engine) cursor.close()
tables = inspector.get_table_names() conn.close()
print("Successfully initialized database with direct SQLite")
except Exception as e:
print(f"Error during direct SQLite initialization: {e}")
import traceback
print(traceback.format_exc())
# Now try with SQLAlchemy as a backup approach
try:
print("Attempting SQLAlchemy database initialization...")
# Try to create all tables from models
Base.metadata.create_all(bind=engine)
print("Successfully created tables with SQLAlchemy")
# Verify tables exist
with engine.connect() as conn:
# Get list of tables
result = conn.execute(text(
"SELECT name FROM sqlite_master WHERE type='table'"
))
tables = [row[0] for row in result]
print(f"Tables in database: {', '.join(tables)}") print(f"Tables in database: {', '.join(tables)}")
if 'task' not in tables: # Verify task table exists
print("WARNING: 'task' table not created!") if 'task' in tables:
else: # Check if task table is empty
print("'task' table successfully created.") result = conn.execute(text("SELECT COUNT(*) FROM task"))
task_count = result.scalar()
print(f"Task table contains {task_count} records")
except Exception as e: # If table exists but is empty, add a sample task
print(f"Error creating tables: {e}") if task_count == 0:
print("Adding sample task with SQLAlchemy")
# Print database info for debugging from app.models.task import Task, TaskPriority, TaskStatus
try: sample_task = Task(
print(f"Database file size: {os.path.getsize(db_file)} bytes") title="Sample SQLAlchemy Task",
print(f"Database file permissions: {oct(os.stat(db_file).st_mode)[-3:]}") description="This is a sample task created with SQLAlchemy",
print(f"Database dir permissions: {oct(os.stat(db_dir).st_mode)[-3:]}") priority=TaskPriority.MEDIUM,
except Exception as e: status=TaskStatus.TODO,
print(f"Error getting file info: {e}") completed=False,
created_at=datetime.utcnow(),
print("Database initialization completed") updated_at=datetime.utcnow()
)
from app.db.session import SessionLocal
db = SessionLocal()
db.add(sample_task)
db.commit()
db.close()
print("Added sample task with SQLAlchemy")
else:
print("WARNING: 'task' table not found!")
print("SQLAlchemy database initialization completed")
except Exception as e: except Exception as e:
print(f"Database initialization error: {str(e)}") print(f"Error during SQLAlchemy initialization: {e}")
print("Continuing anyway...") import traceback
print(traceback.format_exc())
print("Continuing despite SQLAlchemy initialization error...")
def create_test_task(): def create_test_task():

96
app/db/session.py
View File

@ -1,5 +1,6 @@
import os import os
import time import time
import sqlite3
from typing import Generator from typing import Generator
from pathlib import Path from pathlib import Path
@ -9,71 +10,102 @@ from sqlalchemy.exc import OperationalError, SQLAlchemyError
from app.core.config import settings, DB_DIR from app.core.config import settings, DB_DIR
# Ensure the database directory exists # Define database filepath - use DB_DIR from config
db_file = DB_DIR / "db.sqlite" db_file = DB_DIR / "db.sqlite"
print(f"Database file path: {db_file}") print(f"Database file path: {db_file}")
# Configure SQLite connection with more robust settings # Try to touch the database file to ensure it exists
try:
if not db_file.exists():
# Try to create an empty database file
db_file.touch()
print(f"Database file created or verified: {db_file}")
except Exception as e:
print(f"Warning: Could not create database file: {e}")
# Configure SQLite connection with simplified, robust settings
engine = create_engine( engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL, settings.SQLALCHEMY_DATABASE_URL,
connect_args={ connect_args={
"check_same_thread": False, "check_same_thread": False,
"timeout": 30, # Wait up to 30 seconds for the lock "timeout": 30, # Wait up to 30 seconds for the lock
}, },
pool_recycle=3600, # Recycle connections after 1 hour # Minimal pool settings for stability
pool_pre_ping=True, # Verify connections before usage pool_pre_ping=True, # Verify connections before usage
pool_size=10, # Maximum 10 connections echo=True, # Log all SQL for debugging
max_overflow=20, # Allow up to 20 overflow connections
) )
# Add SQLite optimizations # Add essential SQLite optimizations
@event.listens_for(engine, "connect") @event.listens_for(engine, "connect")
def optimize_sqlite_connection(dbapi_connection, connection_record): def optimize_sqlite_connection(dbapi_connection, connection_record):
"""Configure SQLite connection for better performance and reliability.""" """Configure SQLite connection for better reliability."""
# Enable WAL journal mode for better concurrency try:
dbapi_connection.execute("PRAGMA journal_mode=WAL") # These are essential for SQLite stability
# Ensure foreign keys are enforced dbapi_connection.execute("PRAGMA journal_mode=WAL")
dbapi_connection.execute("PRAGMA foreign_keys=ON") dbapi_connection.execute("PRAGMA synchronous=NORMAL")
# Synchronous setting for better performance with decent safety except Exception as e:
dbapi_connection.execute("PRAGMA synchronous=NORMAL") print(f"Warning: Could not configure SQLite connection: {e}")
# Cache settings
dbapi_connection.execute("PRAGMA cache_size=-64000") # 64MB cache
# Temp storage in memory
dbapi_connection.execute("PRAGMA temp_store=MEMORY")
# Simplified Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# More robust database access with retry logic # More robust database access with retry logic and error printing
def get_db() -> Generator: def get_db() -> Generator:
""" """
Get a database session with retry logic for transient errors. Get a database session with retry logic for transient errors.
Creates the database file and tables if they don't exist.
""" """
db = None db = None
retries = 3 retries = 3
retry_delay = 0.5 # Start with 0.5 second delay
for attempt in range(retries): for attempt in range(retries):
try: try:
db = SessionLocal() db = SessionLocal()
# Log connection attempt
print(f"Database connection attempt {attempt+1}")
# Test connection with a simple query # Test connection with a simple query
db.execute("SELECT 1") db.execute("SELECT 1")
# Yield the working connection
# Connection succeeded
print("Database connection successful")
yield db yield db
break break
except (OperationalError, SQLAlchemyError) as e:
except Exception as e:
# Close failed connection
if db: if db:
db.close() db.close()
db = None
# Only retry transient errors like "database is locked" error_msg = f"Database connection attempt {attempt+1} failed: {e}"
if attempt < retries - 1: print(error_msg)
print(f"Database connection attempt {attempt+1} failed: {e}")
print(f"Retrying in {retry_delay} seconds...") # Log critical error details
time.sleep(retry_delay) import traceback
retry_delay *= 2 # Exponential backoff print(f"Error traceback: {traceback.format_exc()}")
else:
print(f"All database connection attempts failed: {e}") # Check if we can directly access database
try:
# Try direct sqlite3 connection as a test
direct_conn = sqlite3.connect(str(db_file))
direct_conn.execute("SELECT 1")
direct_conn.close()
print("Direct SQLite connection succeeded but SQLAlchemy failed")
except Exception as direct_e:
print(f"Direct SQLite connection also failed: {direct_e}")
# Last attempt - raise the error to return 500 status
if attempt == retries - 1:
print("All database connection attempts failed")
raise raise
finally:
# Otherwise sleep and retry
time.sleep(1)
# Always ensure db is closed
try:
if db: if db:
db.close() print("Closing database connection")
db.close()
except Exception as e:
print(f"Error closing database: {e}")

228
main.py
View File

@ -53,16 +53,77 @@ app.add_middleware(
allow_headers=["*"], allow_headers=["*"],
) )
# Add exception handlers for better error reporting # Add comprehensive exception handlers for better error reporting
@app.exception_handler(Exception) @app.exception_handler(Exception)
async def validation_exception_handler(request: Request, exc: Exception): async def global_exception_handler(request: Request, exc: Exception):
import traceback import traceback
import sys
# Log the full error with traceback to stdout/stderr
error_tb = traceback.format_exc()
print(f"CRITICAL ERROR: {str(exc)}", file=sys.stderr)
print(f"Request path: {request.url.path}", file=sys.stderr)
print(f"Traceback:\n{error_tb}", file=sys.stderr)
# Get request info for debugging
headers = dict(request.headers)
# Remove sensitive headers
if 'authorization' in headers:
headers['authorization'] = '[REDACTED]'
if 'cookie' in headers:
headers['cookie'] = '[REDACTED]'
# Include minimal traceback in response for debugging
tb_lines = error_tb.split('\n')
simplified_tb = []
for line in tb_lines:
if line and not line.startswith(' '):
simplified_tb.append(line)
# Create detailed error response
error_detail = { error_detail = {
"detail": f"Internal Server Error: {str(exc)}", "status": "error",
"message": str(exc),
"type": str(type(exc).__name__), "type": str(type(exc).__name__),
"traceback": traceback.format_exc().split("\n") "path": request.url.path,
"method": request.method,
"traceback_summary": simplified_tb[-10:] if len(simplified_tb) > 10 else simplified_tb,
} }
print(f"Error processing request: {error_detail}")
# Add SQLite diagnostic check
try:
import sqlite3
from app.db.session import db_file
# Try basic SQLite operations
conn = sqlite3.connect(str(db_file))
cursor = conn.cursor()
cursor.execute("PRAGMA integrity_check")
integrity = cursor.fetchone()[0]
# Check if task table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task'")
task_table_exists = cursor.fetchone() is not None
# Get file info
import os
file_exists = os.path.exists(db_file)
file_size = os.path.getsize(db_file) if file_exists else 0
# Add SQLite diagnostics to response
error_detail["db_diagnostics"] = {
"file_exists": file_exists,
"file_size": file_size,
"integrity": integrity,
"task_table_exists": task_table_exists,
}
conn.close()
except Exception as db_error:
error_detail["db_diagnostics"] = {"error": str(db_error)}
# Return the error response
print(f"Returning error response: {error_detail}")
return JSONResponse( return JSONResponse(
status_code=500, status_code=500,
content=error_detail, content=error_detail,
@ -103,103 +164,116 @@ def test_db_connection():
""" """
Test database connection and table creation Test database connection and table creation
""" """
from sqlalchemy import text, inspect
from app.db.session import engine
import traceback, os, subprocess
from app.core.config import DB_DIR
try: try:
# Check directory structure and permissions import os
storage_info = { import sqlite3
"app_dir_exists": os.path.exists("/app"), import traceback
"app_dir_writable": os.access("/app", os.W_OK), from sqlalchemy import text
"storage_dir_exists": os.path.exists("/app/storage"), from app.db.session import engine, db_file
"storage_dir_writable": os.access("/app/storage", os.W_OK) if os.path.exists("/app/storage") else False, from app.core.config import DB_DIR
"db_dir_exists": os.path.exists(str(DB_DIR)),
"db_dir_writable": os.access(str(DB_DIR), os.W_OK) if os.path.exists(str(DB_DIR)) else False, # First check direct file access
file_info = {
"db_dir": str(DB_DIR),
"db_file": str(db_file),
"exists": os.path.exists(db_file),
"size": os.path.getsize(db_file) if os.path.exists(db_file) else 0,
"writable": os.access(db_file, os.W_OK) if os.path.exists(db_file) else False,
"dir_writable": os.access(DB_DIR, os.W_OK) if os.path.exists(DB_DIR) else False
} }
# Get disk usage information # Try direct SQLite connection
disk_usage = {} sqlite_test = {}
try: try:
df_output = subprocess.check_output(["df", "-h", "/app/storage"]).decode() conn = sqlite3.connect(str(db_file))
disk_usage["df_output"] = df_output.strip().split('\n') sqlite_test["connection"] = "successful"
# Check if task table exists
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
sqlite_test["tables"] = tables
# Check for task table specifically
if 'task' in tables:
cursor.execute("SELECT COUNT(*) FROM task")
task_count = cursor.fetchone()[0]
sqlite_test["task_count"] = task_count
# Get a sample task if available
if task_count > 0:
cursor.execute("SELECT * FROM task LIMIT 1")
column_names = [description[0] for description in cursor.description]
row = cursor.fetchone()
sample_task = dict(zip(column_names, row))
sqlite_test["sample_task"] = sample_task
# Check database integrity
cursor.execute("PRAGMA integrity_check")
integrity = cursor.fetchone()[0]
sqlite_test["integrity"] = integrity
conn.close()
except Exception as e: except Exception as e:
disk_usage["error"] = str(e) sqlite_test["connection"] = "failed"
sqlite_test["error"] = str(e)
sqlite_test["traceback"] = traceback.format_exc()
# Try database connection # Try SQLAlchemy connection
connection_info = {} sqlalchemy_test = {}
inspector = None
try: try:
with engine.connect() as conn: with engine.connect() as conn:
connection_info["connection"] = "successful" # Basic connectivity test
connection_info["test_query"] = conn.execute(text("SELECT 1")).scalar() result = conn.execute(text("SELECT 1")).scalar()
inspector = inspect(engine) sqlalchemy_test["basic_query"] = result
except Exception as e:
connection_info["connection"] = "failed"
connection_info["error"] = str(e)
# Get table information if connection successful
table_info = {}
if inspector:
tables = inspector.get_table_names()
connection_info["tables"] = tables
for table in tables:
columns = inspector.get_columns(table)
table_info[table] = [col['name'] for col in columns]
# Try to query task table if it exists # Check tables
if 'task' in tables: result = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
with engine.connect() as conn: tables = [row[0] for row in result]
try: sqlalchemy_test["tables"] = tables
task_count = conn.execute(text("SELECT COUNT(*) FROM task")).scalar()
connection_info["task_count"] = task_count # Check task table
except Exception as e: if 'task' in tables:
connection_info["task_query_error"] = str(e) result = conn.execute(text("SELECT COUNT(*) FROM task"))
sqlalchemy_test["task_count"] = result.scalar()
except Exception as e:
sqlalchemy_test["connection"] = "failed"
sqlalchemy_test["error"] = str(e)
sqlalchemy_test["traceback"] = traceback.format_exc()
# Database file information # Check environment
db_file = f"{DB_DIR}/db.sqlite" env_info = {
db_file_info = { "cwd": os.getcwd(),
"path": db_file, "env_variables": {k: v for k, v in os.environ.items() if k.startswith(('DB_', 'SQL', 'PATH'))}
"exists": os.path.exists(db_file),
"size_bytes": os.path.getsize(db_file) if os.path.exists(db_file) else 0,
"writable": os.access(db_file, os.W_OK) if os.path.exists(db_file) else False,
} }
# SQLAlchemy configuration # Try to create a test file
from app.core.config import settings
db_config = {
"sqlalchemy_url": settings.SQLALCHEMY_DATABASE_URL,
"connect_args": {"check_same_thread": False}
}
# Test file creation
write_test = {} write_test = {}
try: try:
test_file = f"{DB_DIR}/test_db.txt" test_path = DB_DIR / "write_test.txt"
with open(test_file, 'w') as f: with open(test_path, 'w') as f:
f.write("Test write access") f.write("Test content")
write_test["success"] = True write_test["success"] = True
write_test["path"] = test_file write_test["path"] = str(test_path)
os.remove(test_file) # Clean up os.unlink(test_path) # Clean up
except Exception as e: except Exception as e:
write_test["success"] = False write_test["success"] = False
write_test["error"] = str(e) write_test["error"] = str(e)
return { return {
"status": "ok", "status": "ok",
"storage_info": storage_info, "timestamp": datetime.utcnow().isoformat(),
"disk_usage": disk_usage, "file_info": file_info,
"connection_info": connection_info, "sqlite_test": sqlite_test,
"table_info": table_info, "sqlalchemy_test": sqlalchemy_test,
"db_file_info": db_file_info, "environment": env_info,
"db_config": db_config,
"write_test": write_test "write_test": write_test
} }
except Exception as e: except Exception as e:
# Catch-all error handler
return { return {
"status": "error", "status": "error",
"global_error": str(e), "message": str(e),
"traceback": traceback.format_exc() "traceback": traceback.format_exc()
} }