Flask-SQLAlchemy check if database server is responsive

Question

I am using flask-SQLAlchemy for my webservice. I would like to have an endpoint that checks status of the utilized MySQL database availability/responsiveness. How would I go about it? Thanks.

Here are relevant pieces of my code:

mywebsvc.py

...
app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://mylogin:mypw@localhost/mydb'

db.init_app(app)
...

models_shared.py

from flask.ext.sqlalchemy import SQLAlchemy

db = SQLAlchemy()

models.py

from models_shared import db

class Car(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    make = db.Column(db.String(64), index=True, unique=True)
    color = db.Column(db.String(64), index=False, unique=False)

routes.py

...
@app.route('/is_available', methods=['GET'])
def is_available():
    #???

Answer 1

There is a fancy library for writing end-point checking condition of the service - healthcheck.

It's useful for asserting that your dependencies are up and running and your application can respond to HTTP requests. The Healthcheck functions are exposed via a user defined flask route so you can use an external monitoring application (Monit, Nagios, Runscope, etc.) to check the status and uptime of your application.

You can use it instead of manually creating end-point because there are some features out of the box (for example EnvironmentDump).

In my application, I had the same need so I implemented check if database is responsive

app = Flask(__name__)

# wrap the flask app and give a heathcheck url
health = HealthCheck(app, "/healthcheck")

def health_database_status():
    is_database_working = True
    output = 'database is ok'

    try:
        # to check database we will execute raw query
        session = DatabaseSession.get_database_session()
        session.execute('SELECT 1')
    except Exception as e:
        output = str(e)
        is_database_working = False

    return is_database_working, output

health.add_check(health_database_status)

As I see, in your application you can execute query with db.engine.execute('SELECT 1').