flask-admin 自定义表单操作并启动后台进程

Question

我是 flask-admin 的新手，想为主要做两件事的视图创建批处理操作：

更新 table 中所选记录的状态，该操作是“运行”
启动运行某些查询、优化等的守护进程或后台进程

编辑： 对于 #2) 我可能想使用 Celery。如果这对于独立问题来说太重要了，我很乐意只关注 #1)，即：我如何简单地更新我选择的记录？看起来超级琐碎，但没有任何效果。

编辑 #2： 我发现这个问题似乎很简单地回答了这个问题，但是我不明白 transaction_service.recalculate_transaction 是什么以及在哪里：

到目前为止，这是我所拥有的，但我一直从该操作中获得 302 状态。所以实际上什么都没有发生。

任何帮助将不胜感激。

__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SECRET_KEY'] = '12345' 
app.config['DEBUG'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:password@localhost/testdb'

db = SQLAlchemy(app)

import test_project.views

views.py

from test_project import app, db
from flask_socketio import SocketIO, emit
from flask import render_template, url_for, redirect
from flask_admin import Admin
from flask_admin.actions import action
from flask_admin.contrib.sqla import ModelView

from test_project.db import TargetTable

socketio = SocketIO(app)
admin = Admin(app, name='Test Tool', template_mode='bootstrap4')


class CustomView(ModelView):
    # Not really using yet.
    pass


class TargetTableAdmin(CustomView):

    form_excluded_columns = ['status']
    column_display_pk = True
    create_modal = True
    can_edit = False
    can_delete = False

    def on_model_change(self, form, model, is_created):
        ###############
        ##### Note: ###  When a record is created the Status is set to "Not Run"
        ###############
        model.status = 'Not Run' #["Running", "Completed", "Failed", "Not Run"]
    
    ################################
    # HOW DO I GET THIS TO WORK?? ##
    ################################
    @action('run', 'Run')    
    def run_target(self, ids):
        query = TargetTable.query.filter(TargetTable.id.in_(ids))
        for target in query.all():
            target.status = 'Running'
            db.session.commit()
            # THIS DOESN'T WORK :(

admin.add_view(TargetTableAdmin(TargetTable, db.session, category='Target'))

db.py

from test_project import app
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

from flask_migrate import Migrate

db = SQLAlchemy(app)
migrate = Migrate(app, db)


class TargetTable(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    start_date = db.Column(db.DateTime, nullable=False)
    end_date = db.Column(db.DateTime, nullable=False)
    status = db.Column(db.String(100), nullable=True)

    def __init__(self, start_date, end_date, status):
        self.start_date = start_date
        self.end_date = end_date
        self.status = status

Answer 1

好的。

所以最终 #1 很容易回答，正如我在上面的评论中提到的那样。我从 __init__.py 文件中导入 db，我应该从 db.py 文件中导入。我不确定是否需要在 __init__.py 文件中初始化数据库，但这是另一个问题。这是修复：

views.py

from test_project import app #### -> remove import here, db 
from flask_socketio import SocketIO, emit
from flask import render_template, url_for, redirect
from flask_admin import Admin
from flask_admin.actions import action
from flask_admin.contrib.sqla import ModelView

from test_project.db import TargetTable, db #### <-- add import here

这变得更难了？回答是因为有很多方法可以给这只猫剥皮。您可以使用几个选项/包：

芹菜/Redis
线程
Flask 执行器
Flask-SocketIO

目前，我正在使用 Flask-SocketIO。具体来说，我在该包中使用了一个名为：start_background_task 的函数。该函数是非阻塞的，一旦调用该函数，页面将一直刷新，而进程在后台运行。这可能会扩展，但我现在不太关心这个，因为目前这只是一个用户最少的内部工具。更新文件：

views.py:

# Imports above ^

socketio = SocketIO(app)
admin = Admin(app, name='Test Tool', template_mode='bootstrap4')


class CustomView(ModelView):
    # Not really using yet.
    pass


def testFunct(record):
    for i in range(10):
        print(i)
        time.sleep(1)
    updateRecord = TargetTable.query.filter( (TargetTable.id == record.id) ).first()
    updateRecord.status = "Completed"
    db.session.commit()


class TargetTableAdmin(CustomView):

    form_excluded_columns = ['status']
    column_display_pk = True
    create_modal = True
    can_edit = False
    can_delete = False

    def on_model_change(self, form, model, is_created):
        ###############
        ##### Note: ###  When a record is created the Status is set to "Not Run"
        ###############
        model.status = 'Not Run' #["Running", "Completed", "Failed", "Not Run"]
    
    ###############
    # IT WORKS!! ##
    ###############
    @action('run', 'Run')    
    def run_target(self, ids):
        query = TargetTable.query.filter(TargetTable.id.in_(ids))
        for t in query.all():
            t.status = 'Running'
            db.session.commit()
            # Kick off background task:
            task = socketio.start_background_task(testFunct, t)

admin.add_view(TargetTableAdmin(TargetTable, db.session, category='Target'))

flask-admin 自定义表单操作并启动后台进程

flask-admin custom form action and kick off background processes

python

sqlalchemy

flask

flask-admin