Flask-SQLAlchemy检查数据库服务器是否响应 [英] Flask-SQLAlchemy check if database server is responsive

问题描述

我正在为我的Web服务使用flask-SQLAlchemy.我希望有一个端点来检查所利用的MySQL数据库可用性/响应性的状态.我将如何处理?谢谢.

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

mywebsvc.py

...
app = Flask(__name__)

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

db.init_app(app)
...

models_shared.py

models_shared.py

from flask.ext.sqlalchemy import SQLAlchemy

db = SQLAlchemy()

models.py

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

routes.py

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

推荐答案

有一个精美的库可用于编写服务的端点检查条件-健康检查.

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

对于断言您的依赖项已启动并正在运行非常有用 并且您的应用程序可以响应HTTP请求.健康检查 通过用户定义的烧瓶路径公开功能，因此您可以使用 外部监控应用程序(Monit，Nagios，Runscope等) 检查您的应用程序的状态和正常运行时间.

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)

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

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

这篇关于Flask-SQLAlchemy检查数据库服务器是否响应的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！