Flask-Admin自定义批量操作更新记录及启动后台进程问题求助

问题修复方案 #

核心问题定位 #

你遇到的状态更新不生效的核心原因是重复初始化了SQLAlchemy实例，两个独立的db实例会话不互通，导致提交的修改没有实际写入数据库：

1. __init__.py中已经实例化了db = SQLAlchemy(app)
2. db.py中又重新实例化了一次db = SQLAlchemy(app)，模型绑定的是这个新的db实例，而view中操作的是顶层的db实例，两边会话不共享，commit自然不生效。

你看到的302状态码是正常现象：Flask-Admin的批量操作执行完成后默认会重定向回列表页，这个不是错误。

修复步骤 #

1. 修正db.py的重复初始化问题

删除db.py中重复的db实例化代码，直接导入顶层已经创建的db：

from test_project import app, db # 直接导入__init__.py中初始化的db
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
from flask_migrate import Migrate

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)

    # 注：SQLAlchemy默认自带构造方法，不需要手动写__init__
    def __init__(self, start_date, end_date, status):
        self.start_date = start_date
        self.end_date = end_date
        self.status = status

2. 优化views.py中的批量更新逻辑

替换原来的循环更新逻辑为批量更新，减少数据库交互，同时添加操作提示：

from flask import flash # 新增导入flash用于显示操作结果提示

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):
        model.status = 'Not Run' 

    @action('run', 'Run', '确定要将选中的任务设置为运行状态吗？') # 第三个参数为操作确认弹窗提示，可选
    def run_target(self, ids):
        # 批量更新，不需要循环查询每条记录
        update_count = TargetTable.query.filter(TargetTable.id.in_(ids)).update({'status': 'Running'})
        db.session.commit()
        flash(f'成功将{update_count}条记录的状态更新为Running')

后台任务实现补充（Celery方案） #

如果要实现第二个后台执行任务的需求，只需要在状态更新完成后调用Celery异步任务即可，示例逻辑：

# 提前定义好你的Celery任务后导入
from your_celery_module import run_optimize_task

@action('run', 'Run', '确定要运行选中的任务吗？')
def run_target(self, ids):
    TargetTable.query.filter(TargetTable.id.in_(ids)).update({'status': 'Running'})
    db.session.commit()
    # 批量提交异步任务
    for task_id in ids:
        run_optimize_task.delay(task_id)
    flash(f'已提交{len(ids)}个任务到后台执行')

你之前看到的transaction_service.recalculate_transaction是其他开发者自己封装的业务方法，不属于Flask-Admin内置能力，直接替换为你自己的业务逻辑即可。

内容的提问来源于stack exchange，提问作者mrp