48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
import time
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.schemas import HealthCheck, HealthResponse
|
|
|
|
router = APIRouter(tags=['health'])
|
|
START_TIME = time.time()
|
|
|
|
|
|
def build_health(db: Session) -> HealthResponse:
|
|
db_check: HealthCheck
|
|
try:
|
|
t0 = time.perf_counter()
|
|
db.execute(text("SELECT 1"))
|
|
latency_ms = round((time.perf_counter() - t0) * 1000, 2)
|
|
db_check = HealthCheck(status='pass', latency_ms=latency_ms, detail='sqlite ok')
|
|
except Exception as exc:
|
|
db_check = HealthCheck(status='fail', detail=str(exc))
|
|
overall = 'pass' if db_check.status == 'pass' else 'fail'
|
|
return HealthResponse(
|
|
status=overall,
|
|
service='netmapper-backend',
|
|
version='0.1.0',
|
|
time=datetime.now(timezone.utc),
|
|
uptime_s=round(time.time() - START_TIME, 2),
|
|
checks={'db': db_check},
|
|
)
|
|
|
|
|
|
@router.get('/live', response_model=HealthResponse)
|
|
def live(db: Session = Depends(get_db)) -> HealthResponse:
|
|
return build_health(db)
|
|
|
|
|
|
@router.get('/ready', response_model=HealthResponse)
|
|
def ready(db: Session = Depends(get_db)) -> HealthResponse:
|
|
return build_health(db)
|
|
|
|
|
|
@router.get('/health', response_model=HealthResponse)
|
|
def health(db: Session = Depends(get_db)) -> HealthResponse:
|
|
return build_health(db)
|