Flask API Monitoring

Set up health check endpoints and uptime monitoring for your Flask application. Simple patterns for production-ready monitoring.

Why Monitor Your Flask App?

Flask applications are lightweight by design, which means they rely heavily on external services -- databases, caches, task queues, and third-party APIs. When any of these fail, your Flask app may still respond to requests but return errors or incomplete data. External monitoring catches these issues before they affect users.

• Gunicorn worker crashes -- A segfault or OOM kill takes down a worker silently; other workers keep serving, masking partial failures
• SQLAlchemy connection issues -- Stale connections after a database restart cause intermittent query errors
• Extension failures -- Flask extensions (Flask-Mail, Flask-Caching) can fail independently without crashing the app
• Memory leaks -- Long-running Flask processes with Gunicorn can accumulate memory until workers are killed

Simple Health Endpoint

from flask import Flask, jsonify
from datetime import datetime

app = Flask(__name__)

@app.route('/health')
def health():
    return jsonify({
        'status': 'healthy',
        'timestamp': datetime.utcnow().isoformat()
    })

With Database Check

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import time

app = Flask(__name__)
db = SQLAlchemy(app)

@app.route('/health')
def health():
    return jsonify({'status': 'healthy'})

@app.route('/ready')
def ready():
    checks = {}
    healthy = True

    # Check database
    try:
        start = time.time()
        db.session.execute('SELECT 1')
        checks['database'] = {
            'status': 'ok',
            'response_time_ms': round((time.time() - start) * 1000, 2)
        }
    except Exception as e:
        healthy = False
        checks['database'] = {
            'status': 'error',
            'message': str(e)
        }

    status_code = 200 if healthy else 503
    return jsonify({
        'status': 'healthy' if healthy else 'unhealthy',
        'timestamp': datetime.utcnow().isoformat(),
        'checks': checks
    }), status_code

Blueprint for Health Checks

# health.py
from flask import Blueprint, jsonify, current_app
from datetime import datetime
import time

health_bp = Blueprint('health', __name__)

@health_bp.route('/health')
def health():
    return jsonify({'status': 'healthy'})

@health_bp.route('/ready')
def ready():
    from extensions import db, redis_client

    checks = {}
    healthy = True

    # Database check
    try:
        start = time.time()
        db.session.execute('SELECT 1')
        checks['database'] = {
            'status': 'ok',
            'response_time_ms': round((time.time() - start) * 1000, 2)
        }
    except Exception as e:
        healthy = False
        checks['database'] = {'status': 'error', 'message': str(e)}

    # Redis check
    try:
        start = time.time()
        redis_client.ping()
        checks['redis'] = {
            'status': 'ok',
            'response_time_ms': round((time.time() - start) * 1000, 2)
        }
    except Exception as e:
        healthy = False
        checks['redis'] = {'status': 'error', 'message': str(e)}

    return jsonify({
        'status': 'healthy' if healthy else 'unhealthy',
        'checks': checks,
        'version': current_app.config.get('VERSION', 'unknown')
    }), 200 if healthy else 503

# app.py
from health import health_bp
app.register_blueprint(health_bp)

Using flask-healthz

# Install
pip install flask-healthz

# app.py
from flask import Flask
from flask_healthz import healthz, HealthError

app = Flask(__name__)

def liveness():
    pass  # Just needs to not raise

def readiness():
    try:
        db.session.execute('SELECT 1')
    except Exception:
        raise HealthError("Database not ready")

app.register_blueprint(healthz, url_prefix="/health")
app.config['HEALTHZ'] = {
    "live": liveness,
    "ready": readiness,
}

Exclude from Logging

# Prevent health checks from cluttering logs
import logging

class HealthCheckFilter(logging.Filter):
    def filter(self, record):
        return '/health' not in record.getMessage()

# Apply to werkzeug logger
logging.getLogger('werkzeug').addFilter(HealthCheckFilter())

Best Practices

• Use blueprints — Keep health checks organized and reusable
• Check all dependencies — Database, cache, external APIs
• Add timeouts — Don't let slow checks hang the endpoint
• Return proper status codes — 200 for healthy, 503 for unhealthy