108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Add project root to Python path for imports in alembic migrations
|
|
project_root = Path(__file__).parent.absolute()
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.api.routers import api_router
|
|
from app.core.config import settings
|
|
from app.db import init_db
|
|
|
|
# Initialize the database on startup
|
|
try:
|
|
init_db.init_db()
|
|
except Exception as e:
|
|
print(f"Error initializing database: {e}")
|
|
# Continue with app startup even if DB init fails, to allow debugging
|
|
|
|
app = FastAPI(title=settings.PROJECT_NAME)
|
|
|
|
# Set all CORS enabled origins - Allow all origins
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Add exception handlers for better error reporting
|
|
@app.exception_handler(Exception)
|
|
async def validation_exception_handler(request: Request, exc: Exception):
|
|
import traceback
|
|
error_detail = {
|
|
"detail": f"Internal Server Error: {str(exc)}",
|
|
"type": str(type(exc).__name__),
|
|
"traceback": traceback.format_exc().split("\n")
|
|
}
|
|
print(f"Error processing request: {error_detail}")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content=error_detail,
|
|
)
|
|
|
|
app.include_router(api_router, prefix=settings.API_V1_STR)
|
|
|
|
|
|
@app.get("/health", tags=["health"])
|
|
def health_check():
|
|
"""
|
|
Health check endpoint to verify the application is running correctly
|
|
"""
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/db-test", tags=["health"])
|
|
def test_db_connection():
|
|
"""
|
|
Test database connection and table creation
|
|
"""
|
|
from sqlalchemy import text, inspect
|
|
from app.db.session import engine
|
|
import traceback
|
|
|
|
try:
|
|
inspector = inspect(engine)
|
|
tables = inspector.get_table_names()
|
|
|
|
table_info = {}
|
|
for table in tables:
|
|
columns = inspector.get_columns(table)
|
|
table_info[table] = [col['name'] for col in columns]
|
|
|
|
with engine.connect() as conn:
|
|
# Try to select from the task table to verify it exists
|
|
if 'task' in tables:
|
|
result = conn.execute(text("SELECT COUNT(*) FROM task")).scalar()
|
|
task_count = result
|
|
else:
|
|
task_count = "Table 'task' not found"
|
|
|
|
# Check DB directory
|
|
import os
|
|
from app.core.config import DB_DIR
|
|
db_path = f"{DB_DIR}/db.sqlite"
|
|
db_exists = os.path.exists(db_path)
|
|
|
|
return {
|
|
"status": "ok",
|
|
"connection": "successful",
|
|
"tables": tables,
|
|
"table_details": table_info,
|
|
"task_count": task_count,
|
|
"db_path": db_path,
|
|
"db_exists": db_exists
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"connection": "failed",
|
|
"error": str(e),
|
|
"traceback": traceback.format_exc()
|
|
} |