Flask App：功能運行時更新進度條

我在Flask中構建了一個相當簡單的WebApp，通過網站的API執行功能。我的用戶使用他們的帳戶URL和API令牌填寫表單;當他們提交表單時，我有一個python腳本，通過API從他們的帳戶導出PDF文件。此功能可能需要很長時間，因此我想在窗體頁面上顯示引導程序進度欄，指示腳本在進程中的位置。我的問題是如何在功能運行時更新進度條？這裏是我正在談論的簡化版本。Flask App：功能運行時更新進度條

views.py：

@app.route ('/export_pdf', methods = ['GET', 'POST']) 
def export_pdf(): 
    form = ExportPDF() 
    if form.validate_on_submit(): 
     try: 
     export_pdfs.main_program(form.account_url.data, 
      form.api_token.data) 
     flash ('PDFs exported') 
     return redirect(url_for('export_pdf')) 
     except TransportException as e: 
     s = e.content 
     result = re.search('<error>(.*)</error>', s) 
     flash('There was an authentication error: ' + result.group(1)) 
     except FailedRequest as e: 
     flash('There was an error: ' + e.error) 
    return render_template('export_pdf.html', title = 'Export PDFs', form = form)

export_pdf.html：

{% extends "base.html" %} 

{% block content %} 
{% include 'flash.html' %} 
<div class="well well-sm"> 
    <h3>Export PDFs</h3> 
    <form class="navbar-form navbar-left" action="" method ="post" name="receipt"> 
    {{form.hidden_tag()}} 
    <br> 
    <div class="control-group{% if form.errors.account_url %} error{% endif %}"> 
     <label class"control-label" for="account_url">Enter Account URL:</label> 
     <div class="controls"> 
     {{ form.account_url(size = 50, class = "span4")}} 
     {% for error in form.errors.account_url %} 
      <span class="help-inline">[{{error}}]</span><br> 
     {% endfor %} 
     </div> 
    </div> 
    <br> 
    <div class="control-group{% if form.errors.api_token %} error{% endif %}"> 
     <label class"control-label" for="api_token">Enter API Token:</label> 
     <div class="controls"> 
     {{ form.api_token(size = 50, class = "span4")}} 
     {% for error in form.errors.api_token %} 
      <span class="help-inline">[{{error}}]</span><br> 
     {% endfor %} 
     </div> 
    </div> 
    <br> 
    <button type="submit" class="btn btn-primary btn-lg">Submit</button> 
    <br> 
    <br> 
    <div class="progress progress-striped active"> 
    <div class="progress-bar" role="progressbar" aria-valuenow="0" aria-valuemin="0" aria-valuemax="100" style="width: 0%"> 
    <span class="sr-only"></span> 
    </div> 
</form> 
</div> 
</div> 
{% endblock %}

和export_pdfs.py：

def main_program(url, token): 
    api_caller = api.TokenClient(url, token) 
    path = os.path.expanduser('~/Desktop/'+url+'_pdfs/') 
    pdfs = list_all(api_caller.pdf.list, 'pdf') 
    total = 0 
    count = 1 
    for pdf in pdfs: 
     total = total + 1 
    for pdf in pdfs: 
     header, body = api_caller.getPDF(pdf_id=int(pdf.pdf_id)) 
     with open('%s.pdf' % (pdf.number), 'wb') as f: 
      f.write(body) 
     count = count + 1 
     if count % 50 == 0: 
      time.sleep(1)

在去年函數I共有PDF文件的數量我將出口，並且在處理過程中持續進行計數。如何將當前進度發送到我的.html文件以適應進度條的'style ='標記？最好以我可以在其他頁面上重複使用相同工具進度條的方式。如果我沒有提供足夠的信息，請告訴我。

來源

2014-06-16 FreshCrichard

我不想給一個編碼答案，但讓我指出你的解決方案。一個傳統的想法是開始一個線程來做PDF導出。線程將進度報告給數據庫表。您的瀏覽器前端執行ajax輪詢以從數據庫獲取進度值。 ajax輪詢的替代方案，您可能需要查看flask-socketio以將進度值向下推送到您的瀏覽器。這種選擇可能需要更多的工程努力。 – chfw

@chfw的想法是你應該如何處理它。但不是一個線程，它應該是一個等待工作的額外過程。而不是一個數據庫，我會使用類似Redis的東西，並通過消息隊列進行通信。最後，我不推薦使用AJAX或WebSockets，而是推薦[SSE]（https://developer.mozilla.org/en-US/docs/Server-sent_events），它更容易設置。 – dAnjou

我在本地主機上運行這個簡單但教育性的Flask SSE實現。要處理GAE中的第三方（用戶上傳）庫：

在您的根路徑中創建一個名爲lib的目錄。
拷貝gevent庫目錄到lib目錄。
這些行添加到您的main.py：
```
import sys 
sys.path.insert(0,'lib') 
```
這就是所有。如果您使用lib目錄從一個子文件夾，使用相對參考：sys.path.insert(0, ../../blablabla/lib')

從http://flask.pocoo.org/snippets/116/

# author: [email protected] 
# 
# Make sure your gevent version is >= 1.0 
import gevent 
from gevent.wsgi import WSGIServer 
from gevent.queue import Queue 

from flask import Flask, Response 

import time 


# SSE "protocol" is described here: http://mzl.la/UPFyxY 
class ServerSentEvent(object): 

    def __init__(self, data): 
     self.data = data 
     self.event = None 
     self.id = None 
     self.desc_map = { 
      self.data : "data", 
      self.event : "event", 
      self.id : "id" 
     } 

    def encode(self): 
     if not self.data: 
      return "" 
     lines = ["%s: %s" % (v, k) 
       for k, v in self.desc_map.iteritems() if k] 

     return "%s\n\n" % "\n".join(lines) 

app = Flask(__name__) 
subscriptions = [] 

# Client code consumes like this. 
@app.route("/") 
def index(): 
    debug_template = """ 
    <html> 
     <head> 
     </head> 
     <body> 
     <h1>Server sent events</h1> 
     <div id="event"></div> 
     <script type="text/javascript"> 

     var eventOutputContainer = document.getElementById("event"); 
     var evtSrc = new EventSource("/subscribe"); 

     evtSrc.onmessage = function(e) { 
      console.log(e.data); 
      eventOutputContainer.innerHTML = e.data; 
     }; 

     </script> 
     </body> 
    </html> 
    """ 
    return(debug_template) 

@app.route("/debug") 
def debug(): 
    return "Currently %d subscriptions" % len(subscriptions) 

@app.route("/publish") 
def publish(): 
    #Dummy data - pick up from request for real data 
    def notify(): 
     msg = str(time.time()) 
     for sub in subscriptions[:]: 
      sub.put(msg) 

    gevent.spawn(notify) 

    return "OK" 

@app.route("/subscribe") 
def subscribe(): 
    def gen(): 
     q = Queue() 
     subscriptions.append(q) 
     try: 
      while True: 
       result = q.get() 
       ev = ServerSentEvent(str(result)) 
       yield ev.encode() 
     except GeneratorExit: # Or maybe use flask signals 
      subscriptions.remove(q) 

    return Response(gen(), mimetype="text/event-stream") 

if __name__ == "__main__": 
    app.debug = True 
    server = WSGIServer(("", 5000), app) 
    server.serve_forever() 
    # Then visit http://localhost:5000 to subscribe 
    # and send messages by visiting http://localhost:5000/publish

來源

2014-11-25 08:44:09 guneysus

正如一些人在意見建議，最簡單的解決方法是在另一個運行導出功能線程，並讓您的客戶端通過另一個請求獲取進度信息。有多種方法來處理這個特定的任務。根據你的需要，你可能會選擇一個或多或少的複雜的。

這裏有一個很（非常）最小的關於如何使用線程做例子：

import random 
import threading 
import time 

from flask import Flask 


class ExportingThread(threading.Thread): 
    def __init__(self): 
     self.progress = 0 
     super().__init__() 

    def run(self): 
     # Your exporting stuff goes here ... 
     for _ in range(10): 
      time.sleep(1) 
      self.progress += 10 


exporting_threads = {} 
app = Flask(__name__) 
app.debug = True 


@app.route('/') 
def index(): 
    global exporting_threads 

    thread_id = random.randint(0, 10000) 
    exporting_threads[thread_id] = ExportingThread() 
    exporting_threads[thread_id].start() 

    return 'task id: #%s' % thread_id 


@app.route('/progress/<int:thread_id>') 
def progress(thread_id): 
    global exporting_threads 

    return str(exporting_threads[thread_id].progress) 


if __name__ == '__main__': 
    app.run()

在指數路徑（/），我們產卵每個出口任務的線程，我們返回一個ID給任務，以便客戶端可以稍後使用進度路線（/ progress/[exporting_thread]）檢索它。導出線程每次認爲合適時更新其進度值。

在客戶端，你會得到這樣的事情（本例使用jQuery的）：

function check_progress(task_id, progress_bar) { 
    function worker() { 
     $.get('progress/' + task_id, function(data) { 
      if (progress < 100) { 
       progress_bar.set_progress(progress) 
       setTimeout(worker, 1000) 
      } 
     }) 
    } 
}

至於說，這個例子非常簡約，你或許應該去一個稍微複雜的方法。通常，我們會將特定線程的進度存儲在數據庫或某種緩存中，這樣我們就不會依賴共享結構，從而避免了我的示例所帶來的大部分內存和併發問題。

Redis（https://redis.io）是一種內存數據庫存儲，通常非常適合這類任務。它很好地集成了Python（https://pypi.python.org/pypi/redis）。

來源

2016-11-25 18:40:34

令人驚歎。謝謝。 –

Flask App：功能運行時更新進度條

回答

相關問題