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

Question

我正在使用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():
    #???

Answer 1

有一个奇特的库,用于编写服务的端点检查条件 - healthcheck.

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

您可以使用它而不是手动创建端点,因为有一些开箱即用的功能(例如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').

Answer 2

您可以尝试向 SQLAlchemy 引擎连接添加连接超时：

\n\n

\n

对于闲置八小时或更长时间的连接，MySQL 具有自动连接关闭行为。要避免出现此问题，请使用 pool_recycle 选项来控制任何连接的最长期限\xe2\x80\xa6

\n

\n\n

然后，您可以检查连接是否超时，以查看数据库是否低于可用性阈值。

\n