python – Flask-SQLAlchemy检查数据库服务器是否响应

2023年1月31日 244次阅读

我正在使用flask-SQLAlchemy作为我的webservice.我想有一个端点,检查使用的
MySQL数据库可用性/响应性的状态.我该怎么办呢？谢谢.

以下是我的代码的相关部分：

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():
    #???

最佳答案 有一个奇特的库用于编写服务的终点检查条件 –
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.

您可以使用它而不是手动创建终点,因为有一些开箱即用的功能(例如EnvironmentDump).

在我的应用程序中,我有同样的需求,所以我实现检查数据库是否响应

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)

如我所见,在您的应用程序中,您可以使用db.engine.execute(‘SELECT 1’)执行查询.

点赞