""" Health check endpoint """ from datetime import datetime from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text import redis.asyncio as redis from app.core.database import get_db from app.core.config import settings from app.schemas.financial import HealthCheckResponse router = APIRouter() @router.get( "/health", response_model=HealthCheckResponse, summary="Health check", description="Check the health status of the API and its dependencies" ) async def health_check(db: AsyncSession = Depends(get_db)): """Health check endpoint""" # Check database db_status = "unhealthy" try: result = await db.execute(text("SELECT 1")) if result.scalar(): db_status = "healthy" except Exception: pass # Check Redis cache cache_status = "unhealthy" try: r = redis.from_url(settings.REDIS_URL) await r.ping() cache_status = "healthy" await r.close() except Exception: pass # Check SEC data availability sec_available = True # Simplified for now # Overall status overall_status = "healthy" if db_status != "healthy" or cache_status != "healthy": overall_status = "degraded" return HealthCheckResponse( status=overall_status, version=settings.APP_VERSION, database=db_status, cache=cache_status, sec_data_available=sec_available, timestamp=datetime.utcnow() )