Python健康检查端点怎么实现

wen python案例 2026-07-10 173

本文目录导读：

Python健康检查端点怎么实现

使用 Flask 框架
使用 FastAPI（异步框架）
Django 实现
轻量级 HTTP 服务器实现
Kubernetes 配置示例
最佳实践建议

我来介绍几种实现Python健康检查端点的方法：

使用 Flask 框架

基础实现

from flask import Flask, jsonify
import psutil
import time
app = Flask(__name__)
@app.route('/health')
def health_check():
    return jsonify({
        'status': 'healthy',
        'timestamp': time.time()
    })
@app.route('/health/ready')
def readiness_check():
    """检查应用是否准备好接收流量"""
    # 这里可以检查数据库连接、缓存等
    try:
        # 模拟检查
        return jsonify({
            'status': 'ready',
            'services': {
                'database': 'connected',
                'cache': 'connected'
            }
        })
    except Exception as e:
        return jsonify({
            'status': 'not ready',
            'error': str(e)
        }), 503
@app.route('/health/live')
def liveness_check():
    """检查应用是否运行"""
    return jsonify({
        'status': 'alive',
        'uptime': time.time()
    })
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

详细健康检查（包含资源监控）

from flask import Flask, jsonify
import psutil
import time
import os
app = Flask(__name__)
class HealthChecker:
    @staticmethod
    def check_disk():
        disk = psutil.disk_usage('/')
        return {
            'total_gb': disk.total / (1024**3),
            'used_gb': disk.used / (1024**3),
            'free_gb': disk.free / (1024**3),
            'percent_used': disk.percent
        }
    @staticmethod
    def check_memory():
        memory = psutil.virtual_memory()
        return {
            'total_gb': memory.total / (1024**3),
            'available_gb': memory.available / (1024**3),
            'percent_used': memory.percent
        }
    @staticmethod
    def check_cpu():
        return {
            'percent': psutil.cpu_percent(interval=1),
            'count': psutil.cpu_count()
        }
@app.route('/health/detailed')
def detailed_health():
    start_time = time.time()
    health_status = {
        'status': 'healthy',
        'timestamp': start_time,
        'application': {
            'name': 'my-app',
            'version': '1.0.0',
            'pid': os.getpid()
        },
        'system': {
            'disk': HealthChecker.check_disk(),
            'memory': HealthChecker.check_memory(),
            'cpu': HealthChecker.check_cpu()
        }
    }
    # 检查是否有任何健康状况不佳
    if health_status['system']['disk']['percent_used'] > 90:
        health_status['status'] = 'degraded'
    health_status['response_time_ms'] = (time.time() - start_time) * 1000
    return jsonify(health_status)
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

使用 FastAPI（异步框架）

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
import psutil
import time
from typing import Dict
app = FastAPI()
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": time.time()
    }
@app.get("/health/ready")
async def readiness_check():
    # 检查依赖服务
    dependencies_healthy = True
    if dependencies_healthy:
        return {"status": "ready"}
    else:
        return JSONResponse(
            status_code=503,
            content={"status": "not_ready"}
        )
@app.get("/health/live")
async def liveness_check():
    return {"status": "alive"}
# 详细检查
@app.get("/health/detailed")
async def detailed_health():
    start_time = time.time()
    health_data = {
        "status": "healthy" if psutil.cpu_percent() < 90 else "degraded",
        "timestamp": start_time,
        "system": {
            "cpu": psutil.cpu_percent(interval=0.5),
            "memory": psutil.virtual_memory()._asdict(),
            "disk": psutil.disk_usage('/')._asdict()
        }
    }
    return health_data
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Django 实现

urls.py

from django.urls import path
from . import views
urlpatterns = [
    path('health/', views.health_check, name='health_check'),
    path('health/ready/', views.readiness_check, name='readiness'),
    path('health/live/', views.liveness_check, name='liveness'),
]

views.py

from django.http import JsonResponse
from django.db import connections
from django.core.cache import cache
import time
def health_check(request):
    return JsonResponse({
        'status': 'healthy',
        'timestamp': time.time()
    })
def readiness_check(request):
    """检查Django应用是否就绪"""
    try:
        # 检查数据库
        db_conn = connections['default']
        db_conn.ensure_connection()
        db_status = 'connected'
    except Exception:
        db_status = 'disconnected'
    # 检查缓存
    try:
        cache.set('_health_check', 'ok', 5)
        cache.get('_health_check')
        cache_status = 'connected'
    except Exception:
        cache_status = 'disconnected'
    if db_status == 'connected' and cache_status == 'connected':
        return JsonResponse({
            'status': 'ready',
            'database': db_status,
            'cache': cache_status
        })
    else:
        return JsonResponse({
            'status': 'not_ready',
            'database': db_status,
            'cache': cache_status
        }, status=503)
def liveness_check(request):
    return JsonResponse({'status': 'alive'})

轻量级 HTTP 服务器实现

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health' or self.path == '/health/live':
            self._send_response({
                'status': 'healthy',
                'timestamp': time.time()
            })
        elif self.path == '/health/ready':
            # 这里进行准备检查
            self._send_response({
                'status': 'ready',
                'database': 'connected'
            })
        else:
            self.send_response(404)
            self.end_headers()
    def _send_response(self, data):
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())
if __name__ == '__main__':
    server = HTTPServer(('0.0.0.0', 8080), HealthHandler)
    print('Health check server running on port 8080...')
    server.serve_forever()

Kubernetes 配置示例

apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: my-app
  template:
    metadata:
      labels:
        app: my-app
    spec:
      containers:
      - name: my-app
        image: my-app:latest
        ports:
        - containerPort: 5000
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 5000
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health/live
            port: 5000
          initialDelaySeconds: 15
          periodSeconds: 20
        startupProbe:
          httpGet:
            path: /health
            port: 5000
          initialDelaySeconds: 3
          periodSeconds: 5

最佳实践建议

区分探针类型：
- 存活检查（Liveness）：检测应用是否运行
- 就绪检查（Readiness）：检测应用是否准备好处理请求
- 启动检查（Startup）：检测应用是否启动完成
返回适当的HTTP状态码：
- 200：健康
- 503：不健康或未就绪
添加必要的依赖检查：
- 数据库连接
- 缓存服务
- 外部API
- 消息队列
添加性能指标：
- 响应时间
- CPU/内存使用率
- 磁盘空间
安全性考虑：
- 限制访问IP
- 添加认证（如果需要）
- 不暴露敏感信息

选择哪种实现方式取决于你的技术栈和需求，Flask/FastAPI适合微服务，Django适合全栈应用,轻量级HTTP服务器适合简单的监控场景。

本文地址： https://www.bmjup.com/post/18986.html