forked from erp-dev/erp
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import logging
|
||
import os
|
||
|
||
from django.db import connections, transaction
|
||
from rest_framework.response import Response
|
||
from rest_framework.views import APIView
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class HealthCheckView(APIView):
|
||
"""
|
||
GET /api/v2/health/
|
||
|
||
仅做最轻量的 DB 查询（SELECT 1）。
|
||
若失败（含超时）：直接退出当前进程，让容器重启。
|
||
"""
|
||
|
||
authentication_classes: list = []
|
||
permission_classes: list = []
|
||
|
||
# 单条查询超时（毫秒）。避免 health check 卡死占用 worker。
|
||
STATEMENT_TIMEOUT_MS = 2000
|
||
|
||
def get(self, request):
|
||
try:
|
||
# 用事务包住 SET LOCAL，确保 statement_timeout 只对本次请求生效
|
||
with transaction.atomic(using="default"):
|
||
with connections["default"].cursor() as cursor:
|
||
cursor.execute("SET LOCAL statement_timeout = %s;", [self.STATEMENT_TIMEOUT_MS])
|
||
cursor.execute("SELECT 1;")
|
||
cursor.fetchone()
|
||
except Exception:
|
||
logger.exception("api_v2 health check failed; exiting process")
|
||
os._exit(1) # 等价于 Go 的 os.Exit(1)：立即退出进程
|
||
|
||
return Response({"status": "ok"})
|
||
|
||
|