ab0e9b973a
- Set up FastAPI project structure with SQLite database - Create database models for tomato images and severity classifications - Implement image upload and processing endpoints - Develop a segmentation model for tomato disease severity detection - Add API endpoints for analysis and results retrieval - Implement health check endpoint - Set up Alembic for database migrations - Update project documentation
81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, status
|
|
from sqlalchemy.orm import Session
|
|
from app.db.session import get_db
|
|
from typing import Dict
|
|
import time
|
|
from pathlib import Path
|
|
from app.core.config import settings
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/health", status_code=status.HTTP_200_OK)
|
|
def health_check(db: Session = Depends(get_db)) -> Dict[str, any]:
|
|
"""
|
|
Perform a health check of the service.
|
|
|
|
Checks:
|
|
- Database connection
|
|
- Storage directories
|
|
- Any critical services the application depends on
|
|
"""
|
|
health_data = {
|
|
"status": "ok",
|
|
"timestamp": time.time(),
|
|
"checks": {
|
|
"database": check_database(db),
|
|
"storage": check_storage(),
|
|
}
|
|
}
|
|
|
|
# If any check failed, update the overall status
|
|
if any(not check["status"] for check in health_data["checks"].values()):
|
|
health_data["status"] = "degraded"
|
|
|
|
return health_data
|
|
|
|
def check_database(db: Session) -> Dict[str, any]:
|
|
"""Check if the database is accessible."""
|
|
try:
|
|
# Simple query to check if the database is responding
|
|
db.execute("SELECT 1")
|
|
return {
|
|
"status": True,
|
|
"message": "Database connection successful"
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"status": False,
|
|
"message": f"Database connection failed: {str(e)}"
|
|
}
|
|
|
|
def check_storage() -> Dict[str, any]:
|
|
"""Check if storage directories are accessible and writable."""
|
|
storage_status = True
|
|
messages = []
|
|
|
|
# Check main storage directories
|
|
for directory in [settings.DB_DIR, settings.IMAGE_DIR, settings.MODEL_DIR]:
|
|
dir_path = Path(directory)
|
|
|
|
if not dir_path.exists():
|
|
try:
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
messages.append(f"Created missing directory: {dir_path}")
|
|
except Exception as e:
|
|
storage_status = False
|
|
messages.append(f"Failed to create directory {dir_path}: {str(e)}")
|
|
continue
|
|
|
|
# Check if directory is writable by trying to create a temp file
|
|
try:
|
|
temp_file = dir_path / ".health_check_temp"
|
|
temp_file.touch()
|
|
temp_file.unlink()
|
|
except Exception as e:
|
|
storage_status = False
|
|
messages.append(f"Directory {dir_path} is not writable: {str(e)}")
|
|
|
|
return {
|
|
"status": storage_status,
|
|
"message": "; ".join(messages) if messages else "All storage directories are accessible and writable"
|
|
} |