我對flask-admin相當陌生，想為一個主要做兩件事的視圖創建一個批處理操作：
- 更新操作為“正在運行”的表中所選記錄的狀態
- 啟動運行一些查詢、優化等的守護行程或后臺行程。
編輯：對于#2）我可能想使用芹菜。如果這對于獨立問題來說太重要了，我很樂意只關注 #1)，即： 如何簡單地更新我選擇的記錄？看起來非常微不足道，但沒有任何效果。
編輯 #2：我發現這個問題似乎相當簡單地回答了這個問題，但是我不明白它是什么以及在哪里transaction_service.recalculate_transaction：Flask Admin 擴展“帶有選擇”-下拉選單與自定義按鈕
到目前為止，這是我所擁有的，但我不斷從操作中獲得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
視圖.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'))
資料庫檔案
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
uj5u.com熱心網友回復：
好的。
- 所以最終#1 很容易回答，正如我在上面的評論中提到的。我正在從檔案中匯入資料庫
__init__.py，我應該從db.py檔案中匯入。我不確定是否需要初始化__init__.py檔案中的資料庫，但這是一個不同的問題。這是修復：
視圖.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-Executor
- Flask-SocketIO
現在，我正在使用 Flask-SocketIO。具體來說，我在該包中使用了一個名為：start_background_task. 該函式是非阻塞的，一旦呼叫該函式，頁面將在行程在后臺運行時一直重繪 。這可能會擴展，但我現在不太擔心，因為這只是目前用戶最少的內部工具。更新后的檔案：
視圖.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'))
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/316762.html
標籤：Python 烧瓶 sqlalchemy 烧瓶管理员