Coverage for polar/health/endpoints.py: 79%

1from fastapi import Depends, HTTPException 

2from redis import RedisError 

3from sqlalchemy import select 

4from sqlalchemy.exc import SQLAlchemyError 

5 

6from polar.postgres import AsyncSession, get_db_session 

7from polar.redis import Redis, get_redis 

8from polar.routing import APIRouter 

9 

10router = APIRouter(tags=["health"], include_in_schema=False) 

11 

12 

13@router.get("/healthz") 

14async def healthz( 

15 session: AsyncSession = Depends(get_db_session), redis: Redis = Depends(get_redis) 

16) -> dict[str, str]: 

17 try: 

18 await session.execute(select(1)) 

19 except SQLAlchemyError as e: 

20 raise HTTPException(status_code=503, detail="Database is not available") from e 

21 

22 try: 

23 await redis.ping() 

24 except RedisError as e: 

25 raise HTTPException(status_code=503, detail="Redis is not available") from e 

26 

27 return {"status": "ok"}