帶你認識 flask 後台作業

• 2019 年 12 月 10 日
• 筆記

01

任務類別簡介

任務進程為後台作業提供了一個便捷的解決方案。Worker過程獨立於應用程式運行，甚至可以位於不同的系統上。應用程式和worker之間的通訊是通過消息完成的。通過與物理相互作用來監視其進度。下圖展示了一個典型的實現：

Python中最流行的任務類別是Celery。這是一個相當複雜的重疊，它有很多選項並支援多個消息示例。另一個流行的Python任務位置是Redis Queue（RQ），它犧牲了一些替代，，僅支援Redis消息本身，但作為交換，它的建立要比Celery簡單長度

Celery和RQ都非常適合在Flask應用程式中支援後台任務，所以我可以選擇更簡單的RQ。不過，用Celery實現相同的功能其實也不難。如果您對Celery更有吸引力，可以閱讀我的部落格中的將Celery與Flask文章一起使用

02

使用RQ

RQ是一個標準的Python三方重疊，用pip安裝：

(venv) $ pip install rq  (venv) $ pip freeze > requirements.txt

正如我前面提到的，應用和RQ worker之間的通訊將在Redis消息中執行，因此你需要運行Redis伺服器。有很多途徑來安裝和運行Redis伺服器，可以下載其內核並執行編譯和安裝。如果你使用的是Windows中，微軟在此處維護了Redis的的安裝程式。在Linux的上，你可以通過作業系統的軟體包管理器安裝Redis的。Mac OS X的用戶可以運行brew install redis，使用然後redis-server命令手動啟動服務

除了確保服務正在運行並可以識別RQ訪問之外，你不需要與Redis進行其他交互

03

創建任務

一個任務，不過是一個Python函數而已。以下是一個示例任務，我將其引入一個新的app / tasks.py模組：

app / tasks.py：示例後台任務

import time    def example(seconds):      print('Starting task')      for i in range(seconds):          print(i)          time.sleep(1)      print('Task completed')

該任務將秒數作為參數，然後在該時間量內等待，並每秒列印一次計數器

04

運行 RQ 工人

任務準備就緒，可以通過rq worker來啟動一個worker進程了：

(venv) $ rq worker microblog-tasks  18:55:06 RQ worker 'rq:worker:miguelsmac.90369' started, version 0.9.1  18:55:06 Cleaning registries for queue: microblog-tasks  18:55:06  18:55:06 *** Listening on microblog-tasks...

microblog-tasks如果您想啟動多個worker來擴展量子，您只需要運行rq worker來生成更多連接到同一個模型的進程，就可以使用Worker進程現在連接到了Redis，並在稱為的上面上查看可能的分配給它的任何作業。在生產環境中，您可能希望至少運行可用的CPU數量的工人。。然後，，當作業出現在特定位置時，任何可用的worker進程都可以獲取它

05

執行任務

現在打開第二個終端窗口並激活虛擬環境。我將使用shell會話來啟動worker中的example()任務：

>>> from redis import Redis  >>> import rq  >>> queue = rq.Queue('microblog-tasks', connection=Redis.from_url('redis://'))  >>> job = queue.enqueue('app.tasks.example', 23)  >>> job.get_id()  'c651de7f-21a8-4068-afd5-8b982a6f6d32'

如果採用的是Redis伺服器運行在不同的主機或埠號上，則使用RQ的Queue類表示從應用程式端看到的任務類型。Redis則需要使用其他URL。

隊列的enqueue()方法用於將作業添加到隊列中。第一個參數是要執行的任務的名稱，可直接傳入函數對象或導入字元串。我發現傳入字元串更加方便，因為不需要在應用程式對enqueue()預期的任何剩餘參數將被傳遞給worker中運行的函數。

enqueue()只要進行了調用，運行著RQ worker的終端窗口上就會出現一些活動。你會看到example()函數正在運行，並且連續列印一次計數器。同時，你的其他終端不會被分開，你可以繼續在shell在上面的示例中，我調用job.get_id()方法來獲取分配給任務的唯一標識符。你可以嘗試使用另一個有趣表達式來檢查worker上的函數是否已完成：

>>> job.is_finished  False

如果你像我在上面的示例中那樣傳遞了23，那麼函數將運行約23秒。在那之後，job.is_finished表達式將True轉化為。就是這麼簡單，炫酷否？

一旦函數完成，worker又回到等待作業的狀態，所以如果你想進行更多的實驗，你可以用不同的參數重複執行enqueue()調用。），但最終會被刪除。這很重要，任務類別不保留已執行作業的歷史記錄

06

報告任務進度

通常，對於長期運行的任務，您需要將一些進度資訊提供給應用程式，從而可以將其顯示給用戶。RQ通過使用作業對象的meta屬性來支援這一點。讓我重新編寫example()任務來編寫進度報告：

app / tasks.py：：帶進度的示例後台任務

import time  from rq import get_current_job    def example(seconds):      job = get_current_job()      print('Starting task')      for i in range(seconds):          job.meta['progress'] = 100.0 * i / seconds          job.save_meta()          print(i)          time.sleep(1)      job.meta['progress'] = 100      job.save_meta()      print('Task completed')

這個新版本的example()使用RQ的get_current_job()函數來獲取一個作業實例，該實例與提交任務時返回給應用程式的實例類似。作業對象的meta屬性是一個字典，任務可以編寫任何想要的與應用程式通訊的自定義數據。在此示例中，我編寫了progress，表示完成任務的百分比。每次進程更新時，我都調用job.save_meta()指示RQ將數據寫入Redis，應用程式可以在其中找到它。

在應用程式方面（目前只是一個Python shell），我可以運行此任務，然後監視進度，如下所示：

>>> job = queue.enqueue('app.tasks.example', 23)  >>> job.meta  {}  >>> job.refresh()  >>> job.meta  {'progress': 13.043478260869565}  >>> job.refresh()  >>> job.meta  {'progress': 69.56521739130434}  >>> job.refresh()  >>> job.meta  {'progress': 100}  >>> job.is_finished  True

如您所見，在另一側，meta屬性可以被重新讀。需要調用refresh()方法來從Redis更新內容

07

任務的資料庫表示

對於Web應用程式，情況會變得更複雜一些，因為一旦任務傳遞請求的處理而啟動，該請求隨即結束，而該任務因為我希望應用程式跟蹤每個用戶正在運行的任務，所以我需要使用資料庫表來維護狀態。你可以在下面看到新的Task模型實現：

app / models.py：任務模型

# ...  import redis  import rq    class User(UserMixin, db.Model):      # ...      tasks = db.relationship('Task', backref='user', lazy='dynamic')    # ...    class Task(db.Model):      id = db.Column(db.String(36), primary_key=True)      name = db.Column(db.String(128), index=True)      description = db.Column(db.String(128))      user_id = db.Column(db.Integer, db.ForeignKey('user.id'))      complete = db.Column(db.Boolean, default=False)        def get_rq_job(self):          try:              rq_job = rq.job.Job.fetch(self.id, connection=current_app.redis)          except (redis.exceptions.RedisError, rq.exceptions.NoSuchJobError):              return None          return rq_job        def get_progress(self):          job = self.get_rq_job()          return job.meta.get('progress', 0) if job is not None else 100

模型這個狀語從句：以前的模型有一個有趣的區別的英文id主鍵欄位的英文字元串類型，而不是整數類型。這是因為對於這個模型，我不會依賴資料庫自己的主鍵生成，而是使用由RQ生成的作業標識符。

該模型將存儲符合任務命名規範的名稱（會傳遞給RQ），適用於向用戶顯示的任務描述，該任務的所屬用戶的關係以及任務是否已完成的布爾值。complete欄位的目的是將正在運行的任務與已完成的任務分開，因為運行中的任務需要特殊處理才能顯示最新進度。

get_rq_job()輔助方法可以用給定的任務ID載入RQ Job實例。的英文這通過Job.fetch()完成的，它會從Redis的存在中的數據中載入Job實例。get_progress()方法建立在get_rq_job()的基礎之上，並返回任務的進度百分比。該方法做一些有趣的假設，如果模型中的作業ID不存在於RQ變數中，則表示作業已完成和數據已過期並已從該中刪除，因此在這種情況下返回的百分比為100。同時，如果job存在，但'meta'屬性中找到進度相關的資訊，那麼可以安全地進行該作業計劃運行，但還沒有啟動，所以在這種情況下進度是0。

改進更改資料庫，需要生成新的遷移，然後升級資料庫：

(venv) $ flask db migrate -m "tasks"  (venv) $ flask db upgrade

新模型也可以添加到shell一部分中，盔甲在shell會話中訪問它時無需導入：

microblog.py：添加任務模型到shell上下文中

from app import create_app, db, cli  from app.models import User, Post, Message, Notification, Task    app = create_app()  cli.register(app)    @app.shell_context_processor  def make_shell_context():      return {'db': db, 'User': User, 'Post': Post, 'Message': Message,              'Notification': Notification, 'Task': Task}

08

將RQ與 Flask 集合在一起

Redis服務的連接URL需要添加到配置中：

class Config(object):      # ...      REDIS_URL = os.environ.get('REDIS_URL') or 'redis://'

與往常一樣，Redis連接URL將來自環境變數，如果該變數未定義，則替換為該服務在當前主機的埠上運行並使用URL。

應用工廠函數將負責初始化Redis和RQ：

app / __ init__.py：整合RQ

# ...  from redis import Redis  import rq    # ...    def create_app(config_class=Config):      # ...      app.redis = Redis.from_url(app.config['REDIS_URL'])      app.task_queue = rq.Queue('microblog-tasks', connection=app.redis)        # ...

app.task_queue將成為提交任務的重量。將附加到應用上會提供很大的便利，因為我可以在應用的任何地方使用current_app.task_queue來訪問它。為了方便應用的任何部分提交或檢查任務，我可以在User模型中創建一些輔助方法：

app / models.py：用戶模型中的任務輔助方法

# ...    class User(UserMixin, db.Model):      # ...        def launch_task(self, name, description, *args, **kwargs):          rq_job = current_app.task_queue.enqueue('app.tasks.' + name, self.id,                                                  *args, **kwargs)          task = Task(id=rq_job.get_id(), name=name, description=description,                      user=self)          db.session.add(task)          return task        def get_tasks_in_progress(self):          return Task.query.filter_by(user=self, complete=False).all()        def get_task_in_progress(self, name):          return Task.query.filter_by(name=name, user=self,                                      complete=False).first()

launch_task()方法是將任務提交到RQ，然後將其添加到資料庫中。name參數是函數名稱，如app / tasks.py中所定義的那樣。提交給RQ時，該函數已app.tasks.預先添加到該名稱中以構建符合規範的函數名稱。description參數是對呈現給用戶的任務的友好描述。對於導出用戶動態的函數，我將名稱設置為export_posts，將描述設置為Exporting posts...。其餘參數將傳遞給任務函數。launch_task()函數首先調用隊列的enqueue()方法來提交作業。返回的作業對象包含由RQ分配的任務ID，因此我可以使用它在我的資料庫中創建相應的Task對象

請注意，launch_task()將新的任務對象添加到會話中，但不會發出提交。替代，最好在更高層次函數中的資料庫會話上進行操作，因為它允許您在替代事務中組合由替代這不是一個嚴格的規則，並且，在本章後面的子函數中也會存在一個例外的提交

get_tasks_in_progress()方法返回該用戶未完成任務的列表。稍後您會看到，我使用此方法在將有關正在運行的任務的資訊渲染到用戶的頁面中

最後，get_task_in_progress()是上一個方法的簡化版本並返回指定的任務。我阻止用戶同時啟動兩個或多個相同類型的任務，因此在啟動任務之前，可以使用此方法來確定前一個任務是否還在運行

09

利用 RQ 任務發送電子郵件

不要認為本節偏離主題，我在上面說過，當後台完成任務完成時，將使用包含所有用戶動態的JSON文件向用戶發送電子郵件。我在第十章中生成的電子郵件功能需要通過兩種方式進行擴展。首先，我需要添加對文件附件的支援，剎車我可以附加JSON文件。串列，send_email()函數總是使用後台執行緒初始化發送電子郵件。當我要從後台任務發送電子郵件時（已經是初步的了），基於執行緒的二級後台任務沒有什麼意義，所以我需要同時支援同步和非同步電子郵件的發送。

幸運的是，Flask-Mail支援附件，所以我需要做的就是擴展send_email()函數的控制項關鍵字參數，然後在Message對象中配置它。選擇在前台發送電子郵件時，我只需要添加一個sync=True的關鍵字參數即可：

app / email.py：發送帶附件的郵件

# ...    def send_email(subject, sender, recipients, text_body, html_body,                 attachments=None, sync=False):      msg = Message(subject, sender=sender, recipients=recipients)      msg.body = text_body      msg.html = html_body      if attachments:          for attachment in attachments:              msg.attach(*attachment)      if sync:          mail.send(msg)      else:          Thread(target=send_async_email,              args=(current_app._get_current_object(), msg)).start()

消息類的attach()方法接受三個定義附件的參數：文件名，媒體類型和實際文件數據。文件名就是收件人看到的與附件關聯的名稱。媒體類型定義了這種附件的類型，這有助於電子郵件讀者適當地渲染它。例如，如果您發送為image/png媒體類型，則電子郵件閱讀器會知道該附件是一個影像，在這種情況下，它可以顯示它。對於用戶動態數據文件，我將使用JSON格式，該格式使用application/json媒體類型。最後一個參數包含附件內容的字元串或位元組序列。

簡單來說，send_email()的attachments參數將成為一個元組列表，每個元組將有三個元素對應於attach()的三個參數。因此，我需要轉換列表中的每個元素作為參數發送給attach()。在Python中，如果你想將列表或元組中的每個元素作為參數傳遞給函數，你可以使用func(*args)將這個列表或元祖解包成函數中的多個參數，而不必枯燥地一個個地傳遞，如func(args[0], args[1], args[2])。例如，如果如果沒有，調用將會引發一個參數，即列表。你有一個列表args = [1, 'foo']，func(*args)將會傳遞兩個參數，就和你調用func(1, 'foo')一樣。*args

如電子郵件的同步發送，我需要做的就是，當sync是True的時候恢復成調用mail.send(msg)

10

任務助手

儘管我上面使用的example()任務是一個簡單的獨立函數，但已添加用戶動態的函數卻需要應用中具有的某些功能，例如訪問資料庫和發送電子郵件。因為這將在單獨的進程中運行，所以我需要初始化Flask-SQLAlchemy和Flask-Mail，而Flask-Mail又需要Flask應用程式實例以從中獲取它們的配置。因此，我將在app / tasks.py模組的頂部添加Flask應用程式實例和應用程式：

app / tasks.py：創建應用及其自身

from app import create_app    app = create_app()  app.app_context().push()

當使用flask命令時，根目錄中的microblog.py模組創建應用實例，但RQ worker實際上卻一無所知，所以當任務函數時，應用程式在此模組中創建，因為這是RQ worker要導入的唯一模組。你已經在好幾個地方app.app_context()看到了方法，按下一個使其使應用成為「當前」的應用實例，這樣一來Flask-SQLAlchemy等插件才可以使用current_app.config獲取它們的配置。。根本沒有，current_app表達式會返回一個錯誤。

然後我開始考慮如何在這個函數運行時報告進度。另外通過job.meta字典傳遞進度資訊之外，我還想將通知推送給客戶端，剎車自動動態更新完成百分比。逐步，我將使用我在第二十一章中生成的通知機制。更新將以與未讀消息徽章非常類似的方式工作。當伺服器渲染模板時，則包含從job.meta獲得的「靜態」進度資訊，但一旦頁面置於客戶端的瀏覽器中，通知將使用通知來動態更新百分比。由於通知的原因，更新正在運行的任務的進度將比上一個示例中的操作稍微多一些，所以我將創建一個專用於更新進度的包裝函數：

app / tasks.py：設置任務進度

from rq import get_current_job  from app import db  from app.models import Task    # ...    def _set_task_progress(progress):      job = get_current_job()      if job:          job.meta['progress'] = progress          job.save_meta()          task = Task.query.get(job.get_id())          task.user.add_notification('task_progress', {'task_id': job.get_id(),                                                       'progress': progress})          if progress >= 100:              task.complete = True          db.session.commit()

任務導出可以調用_set_task_progress()來記錄進度百分比。函數該首先將百分比寫入job.meta字典搜索並將其保存到Redis的，然後從資料庫載入相應的任務對象，使用並task.user已有的add_notification()方法將通知推送給請求該任務的用戶。通知將被命名為task_progress，並且伴隨關聯的數據將成為具有兩個關聯的字典：任務標識符和進度數值。稍後我將添加JavaScript程式碼來處理這種新的通知類型

該函數查看進度來確認任務函數是否已完成，並在這種情況下下更新資料庫中任務對象的complete屬性。資料庫提交調用通過add_notification()添加的任務和通知對象都立即保存到資料庫。任務，確保不執行任何資料庫更改，因為執行本次調用父父的更改也寫入資料庫

11

實現導出任務

現在所有的準備工作已經完成，可以開始編寫導出函數了。這個函數的高層結構如下：

app / tasks.py：導出用戶動態通用結構

def export_posts(user_id):      try:          # read user posts from database          # send email with data to user      except:          # handle unexpected errors

請求處理器中的應用程式可以防止意外錯誤，因為Flask自身捕獲異常，然後將其整個任務包裝在try / except中。將運行在由RQ控制的單獨前進中，而不是燒瓶，因此如果發生任何意外錯誤，任務將中止，RQ將向控制台顯示錯誤，然後返回等待新的作業。worker的輸出或將其記錄到文件中，否則將永遠不會發現有錯誤。

讓我們從上面帶有注釋的三部分中最簡單的錯誤處理部分開始梳理：

app / tasks.py：更新用戶動態錯誤處理

import sys  # ...    def export_posts(user_id):      try:          # ...      except:          _set_task_progress(100)          app.logger.error('Unhandled exception', exc_info=sys.exc_info())

發生意外錯誤時，我將通過將進度設置為100％來將任務標記為完成，然後使用Flask應用程式中的日誌記錄器對象記錄錯誤以及如何跟蹤資訊（調用sys.exc_info()來獲得）。記錄器來記錄錯誤的好處在於，你可以觀察到你為瓶應用實現的任何日誌記錄機制。例如，在第七章中，我配置了要發送到管理員電子郵件地址的錯誤。只要使用app.logger，我也可以得到這些錯誤資訊

接下來，我將編寫實際的起始程式碼，它只需發出一個資料庫查詢並在循環中遍歷結果，隨之而來的累積在字典中：

app / tasks.py：從資料庫讀取用戶動態

import time  from app.models import User, Post    # ...    def export_posts(user_id):      try:          user = User.query.get(user_id)          _set_task_progress(0)          data = []          i = 0          total_posts = user.posts.count()          for post in user.posts.order_by(Post.timestamp.asc()):              data.append({'body': post.body,                           'timestamp': post.timestamp.isoformat() + 'Z'})              time.sleep(5)              i += 1              _set_task_progress(100 * i // total_posts)            # send email with data to user      except:          # ...

時間格式將採用ISO 8601標準。我使用的Python的datetime對象不存儲時區，因此在以ISO格式導出時間後，我添加了'Z'，它表示UTC

我維護了一個計數器i，並且在進入循環之前還需要發出一個額外的資料庫查詢，查詢total_posts導致用戶動態的總數。使用了i和total_posts，在每個循環迭代我都可以使用從0到100的數字來更新任務進度

您可能會好奇我為什麼會在每個循環time.sleep(5)迭代中加入調用。最終是我想要延長增量所需的時間，剎車在用戶動態不多的情況下也可以方便地查看到逐步進度的增長

下面是函數的最後部分，將會帶上data附件發送郵件給用戶：

app / tasks.py：發送帶用戶動態的郵件給用戶

import json  from flask import render_template  from app.email import send_email    # ...    def export_posts(user_id):      try:          # ...            send_email('[Microblog] Your blog posts',                  sender=app.config['ADMINS'][0], recipients=[user.email],                  text_body=render_template('email/export_posts.txt', user=user),                  html_body=render_template('email/export_posts.html', user=user),                  attachments=[('posts.json', 'application/json',                                json.dumps({'posts': data}, indent=4))],                  sync=True)      except:          # ...

只是其實對send_email()函數的調用。附件被定義為一個元組，其中有三個元素被傳遞給瓶郵件的Message對象的attach()方法。元組中的第三個元素是附件內容，它是用Python中的json.dumps()函數生成的。

這裡引用了一對新模板，它們以純文本和HTML格式提供電子郵件正文的內容。這是文本模板的內容：

app / templates / email / export_posts.txt：更新用戶動態文本郵件模板

Dear {{ user.username }},    Please find attached the archive of your posts that you requested.    Sincerely,    The Microblog Team

這是HTML版本的郵件模板：

app / templates / email / export_posts.html：更新用戶動態HTML郵件模板

<p>Dear {{ user.username }},</p>  <p>Please find attached the archive of your posts that you requested.</p>  <p>Sincerely,</p>  <p>The Microblog Team</p>

12

應用中的導出功能

剩下的就是將這個功能連接到應用，剎車用戶發起請求並通過電子郵件發送用戶動態給他們

下面是新的export_posts視圖函數：

app / main / routes.py：導出用戶動態路由和視圖函數

@bp.route('/export_posts')  @login_required  def export_posts():      if current_user.get_task_in_progress('export_posts'):          flash(_('An export task is currently in progress'))      else:          current_user.launch_task('export_posts', _('Exporting posts...'))          db.session.commit()      return redirect(url_for('main.user', username=current_user.username))

該功能首先檢查用戶是否有未完成的任務，並在這種情況下只是閃現消息。對同一用戶同時執行兩個發起任務是沒有意義的，可以避免。我可以使用前面實現的get_task_in_progress()方法來檢查這種情況

如果一個用戶沒有正在運行的導出任務，則調用launch_task()來啟動它。第一個參數是將傳遞給RQ worker的函數的名稱，改為為app.tasks.。第二個參數只是一個友好的文本描述，將會顯示給用戶。這兩個值都會被寫入資料庫中的任務對象。該函數以重定向到用戶個人主頁結束

我認為最合適的地方是在用戶個人主頁，只有在用戶查看他們自己的主頁時，鏈接在「編輯個人資料」鏈接下面顯示：

app / templates / user.html：用戶個人主頁的導出鏈接

...  <p>      <a href="{{ url_for('main.edit_profile') }}">          {{ _('Edit your profile') }}      </a>  </p>  {% if not current_user.get_task_in_progress('export_posts') %}  <p>      <a href="{{ url_for('main.export_posts') }}">          {{ _('Export your posts') }}      </a>  </p>  ...  {% endif %}

此鏈接的渲染是有條件的，因為我不希望它在用戶已經有發起任務執行時出現。

如果你想嘗試一下，你可以按如下方式啟動應用和RQ worker：

• 確保Redis正在運行
• ：一個終端窗口，啟動至少一個RQ worker實例。本處你可以運行命令rq worker microblog-tasks
• 再打開另一個終端窗口，使用flask run （記得先設置 FLASK_APP變數）命令啟動Flask應用

13

進度通知

為了完善這個功能，我想在後台任務運行時提醒用戶任務完成的進度。在瀏覽Bootstrap組件選項時，我決定在導航欄的下方使用一個Alert組件。橫條。我用藍色的警報框來渲染閃現的消息。現在我要添加一個綠色的警報框來顯示任務進度。樣式如下：

app / templates / base.html：基礎模板中的導出進度Alert組件

...  {% block content %}      <div class="container">          {% if current_user.is_authenticated %}          {% with tasks = current_user.get_tasks_in_progress() %}          {% if tasks %}              {% for task in tasks %}              <div class="alert alert-success" role="alert">                  {{ task.description }}                  <span id="{{ task.id }}-progress">{{ task.get_progress() }}</span>%              </div>              {% endfor %}          {% endif %}          {% endwith %}          {% endif %}          ...  {% endblock %}  ...

外部條件在用戶未登錄時跳過所有與Alert相關的標記。而對於已登錄的用戶，我通過稱為創建的get_tasks_in_progress()方法來獲取當前的任務列表。在當前版本的應用中，我最多只能得到一個結果，因為我可以多個替換任務同時執行，但將來我可能要支援可以共存的其他類型的任務，所以以通用的方式渲染Alert可以節省我以後的時間。

對於每項任務，我都會在頁面上渲染一個警報元素。警報的顏色由第二個CSS樣式控制，本處是alert-success，而在閃現消息是alert-info，引導文檔所有遊戲有關警報的HTML結構的詳細資訊。警報文本包括存儲在Task模型中的description細分，後面跟著完成百分比。

被百分比封裝在具有id屬性的<span>元素中。原因是我要在收到通知時用的JavaScript刷新百分比。我給任務ID附加末尾-progress來構造id屬性。當有通知到達時，通過其中的任務ID，我可以很容易地使用#<task.id>-progress選擇器找到正確的<span>元素來更新。

如果您此時進行嘗試，則每次導航到新頁面時都會看到「靜態」的進度更新。您可以注意到，在啟動導出任務後，您可以自由導航到應用程式的不同頁面，正在運行的任務的狀態始終都會展示出來

為了對span>元素的百分比的動態更新做準備，我將在JavaScript端編寫一個輔助函數：

app / templates / base.html：動態更新任務進度的輔助函數

...  {% block scripts %}      ...      <script>          ...          function set_task_progress(task_id, progress) {              $('#' + task_id + '-progress').text(progress);          }  </script>      ...  {% endblock %}

這個函數接受一個任務id和一個進度值，並使用jQuery為這個任務定位<span>元素，轉換為新進度作為其內容寫入。無需驗證頁面上是否存在該元素，因為如果沒有找到該元素，jQuery將不會執行任何操作。

app / tasks.py中的_set_task_progress()函數每次更新進度時調用add_notification()，就會產生新的通知。而我在第二十一章明智地以完全通用的方式實現了通知功能。所以當瀏覽器定期向伺服器發送時通知更新請求時，瀏覽器會獲得通過add_notification()方法添加的任何通知

但是，這些JavaScript程式碼只能識別具有unread_message_count名稱的那些通知，並忽略其餘部分。我現在需要做的是擴展該函數，通過調用我上面定義的set_task_progress()函數來處理task_progress通知。以下是處理通知更新版本JavaScript程式碼：

app / templates / base.html：通知處理器

for (var i = 0; i < notifications.length; i++) {      switch (notifications[i].name) {          case 'unread_message_count':              set_message_count(notifications[i].data);              break;          case 'task_progress':              set_task_progress(                  notifications[i].data.task_id,                  notifications[i].data.progress);              break;      }      since = notifications[i].timestamp;  }

現在我需要處理兩個不同的通知，我決定用一個switch語句替換檢查unread_message_count通知名稱的if語句，該語句包含我現在需要支援的每個通知。如果你對「 C」系列語言不熟悉，就可能從未見過是switch語句，它提供了一種方便的語法，可以替代一長串的if/elseif語句。這是一個很棒的特性，因為當我需要支援更多通知時，只需簡單地添加case塊即可。

回顧一下，RQ任務附加到task_progress通知的數據是一個包含兩個元素task_id和progress的字典，這兩個元素是我調用set_task_progress()的兩個參數。

如果您現在運行該應用，則綠色Alert插入的進度指示器將每10秒刷新一次（因為刷新通知的時間間隔是10秒）。

如果您要維護非英語語言文件，則需要使用Flask-Babel刷新翻譯文件，然後添加新的翻譯：

(venv) $ flask translate update

如果您使用的是編碼翻譯，那麼我已經為你完成了翻譯工作，因此可以從下載包中提取app / translations / es / LC_MESSAGES / messages.po文件，從而將其添加到您的項目中。

翻譯文件到位後，還要編譯翻譯文件：

(venv) $ flask translate compile