38 lines
1.1 KiB
Django/Jinja
38 lines
1.1 KiB
Django/Jinja
{% if use_dishka and database == 'postgres' -%}
|
|
from dishka.integrations.fastapi import DishkaRoute, FromDishka
|
|
from fastapi import APIRouter
|
|
from sqlalchemy import text
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
router = APIRouter(tags=["health"], route_class=DishkaRoute)
|
|
|
|
|
|
@router.get("")
|
|
@router.get("/")
|
|
async def health(session: FromDishka[AsyncSession]) -> dict[str, bool]:
|
|
result = await session.scalars(text("SELECT 1"))
|
|
return {"db": result.first() == 1}
|
|
{% elif use_dishka and database == 'mongo' -%}
|
|
from dishka.integrations.fastapi import DishkaRoute, FromDishka
|
|
from fastapi import APIRouter
|
|
from pymongo import AsyncMongoClient
|
|
|
|
router = APIRouter(tags=["health"], route_class=DishkaRoute)
|
|
|
|
|
|
@router.get("")
|
|
@router.get("/")
|
|
async def health(client: FromDishka[AsyncMongoClient]) -> dict[str, bool]:
|
|
await client.admin.command("ping")
|
|
return {"db": True}
|
|
{% else -%}
|
|
from fastapi import APIRouter
|
|
|
|
router = APIRouter(tags=["health"])
|
|
|
|
|
|
@router.get("")
|
|
@router.get("/")
|
|
async def health() -> dict[str, bool]:
|
|
return {"ok": True}
|
|
{% endif -%} |