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

发布时间：2020-12-20 11:43:04 所属栏目：Python 来源：网络整理

导读：我正在使用flask-SQLAlchemy作为我的webservice.我想有一个端点,检查使用的 MySQL数据库可用性/响应性的状态.我该怎么办呢？谢谢. 以下是我的代码的相关部分： mywebsvc.py ...app = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://mylog

我正在使用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’)执行查询.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!