-
Notifications
You must be signed in to change notification settings - Fork 3
/
health.py
58 lines (49 loc) · 1.57 KB
/
health.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import time
import fastapi
import sqlalchemy as sa
from fastapi import responses
from sqlalchemy.ext.asyncio import AsyncSession
from app import config, database, schemas
router = fastapi.APIRouter()
class HealthResponse(responses.JSONResponse):
media_type = "application/health+json"
@router.get(
"",
response_class=HealthResponse,
response_model=schemas.Health,
)
async def get_health(
response: HealthResponse,
settings: config.Settings = fastapi.Depends(config.get_settings),
db: AsyncSession = fastapi.Depends(database.get_db),
) -> schemas.Health:
response.headers["Cache-Control"] = "max-age=3600"
stmt = sa.text("SELECT now(), pg_postmaster_start_time()")
t0 = time.perf_counter_ns()
result = await db.execute(stmt)
response_time_ns = time.perf_counter_ns() - t0
now, start = result.one()
content = {
"status": schemas.Status.PASS,
"version": settings.version,
"releaseId": settings.releaseId,
"checks": {
"postgresql:responseTime": [
{
"componentType": "datastore",
"observedValue": response_time_ns / 1000000,
"observedUnit ": "ms",
"time": now,
}
],
"postgresql:uptime": [
{
"componentType": "datastore",
"observedValue": now - start,
"observedUnit ": "s",
"time": now,
}
],
},
}
return schemas.Health(**content)