FLASK：如何与mysql服务器建立连接？答案

【问题标题】：FLASK: How to establish connection with mysql server?FLASK：如何与mysql服务器建立连接？
【发布时间】：2021-06-30 05:53:39
【问题描述】：

我已经有一个来自 Flask 的 mysql 连接，如下所示：

app.config['MYSQL_HOST'] = 'one.hostname.net'
app.config['MYSQL_USER'] = 'my_username'
app.config['MYSQL_PASSWORD'] = 'my_password'
app.config['MYSQL_DB'] = 'user_mydb'

mysql = MySQL(app)

通过此设置，我可以在烧瓶中使用 mysql 数据库连接。但是当涉及到 celery 任务时，它与 flask 位于同一个 python 文件中。

@mycelery.task(bind=True, name='mytask')
def mytask(self, userid, port):
    cursor = mysql.connection.cursor(MySQLdb.cursors.DictCursor)
    cursor.execute('SELECT * FROM mytable WHERE id = %s', (userid,))

它给我一个错误提示

cursor = mysql.connection.cursor(MySQLdb.cursors.DictCursor)
AttributeError: 'NoneType' object has no attribute 'cursor'

我知道这是因为 celery 没有建立 MySQL 连接。但是如何建立连接？这样我就不必在创建任务时连接 MySQL 服务器，就像我们已经通过 MySQL = MySQL(app)?? 建立连接的烧瓶一样

这是我的 celery 设置，如果在此代码中添加一些内容会有所帮助

mycelery = Celery(app.name)
mycelery.conf.update({
    'broker_url': 'filesystem://',
    'broker_transport_options': {
        'data_folder_in': 'app/broker/out',
        'data_folder_out': 'app/broker/out',
        'data_folder_processed': 'app/broker/processed'
    },

    'result_persistent': False,
    'task_serializer': 'json',
    'result_serializer': 'json',
    'accept_content': ['json']})

【问题讨论】：

标签： mysql flask celery

【解决方案1】：

首先，如果你想做数据库操作，你可以在 celery 任务中进行，但你不必将 db 与 celery 连接。您可以将flask与db连接起来，并在您的项目中安装celery并在您的celery任务中进行db操作。

示例：

app.py

from flask import Flask
from flaskext.mysql import MySQL

app = Flask(__name__)
mysql = MySQL()
app.config['MYSQL_DATABASE_USER'] = ''
app.config['MYSQL_DATABASE_PASSWORD'] = ''
app.config['MYSQL_DATABASE_DB'] = ''
app.config['MYSQL_DATABASE_HOST'] = ''
mysql.init_app(app)

tasks.py

@celery.task
def db_connect_things(): 
    conn = mysql.connect()
    cursor =conn.cursor()
    sql_query = """select from where """
    cursor.execute(sql_query)
    ...

celery_config.py

from celery import Celery
celery = Celery(__name__)
celery = Celery('tasks', broker=) # rabbit,redis, ..
celery.conf.update({'CELERY_ACCEPT_CONTENT': ['pickle', 'json', 'msgpack', 'yaml']})
celery.conf.add_defaults(...)

celery.conf.update(CELERYBEAT_SCHEDULE={
    'db_connect_things': {
        'task': 'application.lib.tasks.db_connect_things',
        'schedule': crontab(minute=0, hour='*/12'),
    }})

class ContextTask(celery.Task):
...

【讨论】：