Trong phần này, chúng ta sẽ tìm hiểu cách xây dựng các tác vụ nền có thể được thực thi độc lập với phần mềm máy chủ Web.

Để giúp cho bạn dễ theo dõi, sau đây là danh sách các bài viết trong loạt bài hướng dẫn này:

Bạn có thể truy cập mã nguồn cho phần này tại GitHub.

Phần này sẽ tập trung vào các kỹ thuật cần thiết để xây dựng các tiến trình phức tạp và đòi hỏi thời gian thực thi dài trong ứng dụng. Các tiến trình này không thể thi hành đồng bộ với các yêu cầu từ chương trình khách (client) bởi vì quá trình thực thi chúng sẽ kéo dài và sẽ không cho phép bất kỳ hồi đáp nào từ server về phía client trong suốt thời gian thực thi. Trong Phần 10, chúng ta đã có dịp tìm hiểu một ít về chủ đề này khi chuyển đổi hàm gởi email thành một luồng (thread) chạy ở chế độ nền (background) để người sử dụng không phải chờ 3-4 giây mỗi khi gởi một email. Tuy vậy, dù giải pháp này hoạt động tốt cho tác vụ gởi email, nó lại không hiệu quả lắm khi các tiến trình đòi hỏi thời gian thực thi dài hơn. Trong thực tế, giải pháp được sử dụng rộng rãi hơn là đưa các tác vụ có thời gian thực thi dài vào các tiến trình phục vụ (worker process).

Để minh họa cho nhu cầu về các tiến trình có thời gian thực thi dài, chúng ta sẽ tìm hiểu và xây dựng chức năng xuất file (export). Chức năng này sẽ cho phép một user nhận được một file dữ liệu có chứa tất cả các bài viết của họ. Khi một user sử dụng chức năng này, ứng dụng sẽ khởi động tác vụ xuất file để tạo ra một file JSON có chứa toàn bộ các bài viết của user đó và gởi đến họ qua email. Tất cả hoạt động này sẽ được diễn ra trong một tiến trình phục vụ, đồng thời hiển thị tiến độ công việc qua một thông báo để user có thể quan sát được mức độ hoàn thành theo tỉ lệ phần trăm.

Giới thiệu về Hàng đợi tác vụ (Task Queue)

Hàng đợi tác vụ là một giải pháp tiện lợi cho ứng dụng để thực thi một tác vụ nhờ vào một tiến trình phục vụ. Tiến trình phục vụ được thực thi độc lập với ứng dụng và thậm chí có thể được đặt trên một hệ thống khác. Các giao tiếp giữa ứng dụng và tiến trình phục vụ sẽ được thực hiện qua hàng đợi thông điệp (message queue). Ứng dụng sẽ gởi công việc đến hàng đợi và theo dõi tiến trình công việc bằng cách tương tác với hàng đợi như sơ đồ sau:

Ứng dụng phổ biến nhất cho hàng đợi tác vụ trong Python hiện nay là Celery. Đây là một ứng dụng tương đối phức tạp với nhiều tùy chọn và hỗ trợ một số kiểu hàng đợi khác nhau. Một ứng dụng hàng đợi Python phổ biến khác là Redis Queue (thường được gọi tắt là RQ) và không linh hoạt bằng Celery (như là chỉ hỗ trợ hàng đợi thông điệp của Redis), nhưng lại dễ sử dụng hơn rất nhiều.

Cả Celery lẫn RQ đều có thể hỗ trợ tốt cho các tác vụ nền trong ứng dụng Flask, vì vậy, chúng ta sẽ sử dụng RQ cho ứng dụng của chúng ta vì tính đơn giản của nó. Tuy vậy, việc xây dựng chức năng này với Celery cũng không phải quá khó. Nếu bạn thích dùng Celery hơn RQ, chúng ta có thể trở lại với chủ đề này trong một bài viết khác.

Sử dụng RQ

RQ là một gói Python chuẩn và có thể được cài đặt bằng pip:

(myenv) $ pip3 install rq
(myenv) $ pip3 freeze > requirements.txt

1 2	(myenv) $ pip3 install rq (myenv) $ pip3 freeze > requirements.txt

Như đã nói ở trên, liên lạc giữa ứng dụng và các tiến trình phục vụ của RQ sẽ được thực hiện qua một hàng đợi thông điệp của Redis, vì vậy bạn phải thiết lập một Redis server. Chúng ta có nhiều tùy chọn khác nhau để cài đặt và thực thi một server Redis như là dùng chương trình cài đặt hoặc tải về mã nguồn và biên dịch chúng trên hệ thống của bạn. Nếu sử dụng Windows, bạn có thể tải trình cài đặt Redis server ở đây. Trên Linux, bạn có thể cài đặt thông qua các tiện ích quản lý gói tương ứng với bản phân phối Linux bạn đang sử dụng. Với Mac OS X, bạn có thể dùng lệnh brew install redis (sau khi đã cài homebrew) và khởi động redis thủ công với lệnh redis-server.

Trong các hệ thống sử dụng Debian hoặc Ubuntu, bạn có thể cài đặt phần mềm Redis với các lệnh sau:

$ sudo apt -y update # Cập nhật thông tin về các package trong hệ thống
$ sudo apt -y install redis-server # Cài đặt Redis Server
$ sudo systemctl enable --now redis-server.service # Khởi động và cho phép Redis Server tự động khởi động khi boot.

$ sudo apt -y update # Cập nhật thông tin về các package trong hệ thống

$ sudo apt -y install redis-server # Cài đặt Redis Server

$ sudo systemctl enable --now redis-server.service # Khởi động và cho phép Redis Server tự động khởi động khi boot.

Sau khi khởi động, bạn chỉ cần đảm bảo rằng server đang hoạt động và cho phép truy cập RQ, ngoài ra bạn sẽ không cần phải làm gì thêm.

Cần lưu ý rằng RQ không chạy bằng trình thông dịch Python cho Windows. Nếu bạn đang sử dụng Windows, bạn cần thực thi RQ trong một môi trường mô phỏng Unix. Bạn có thể làm điều này bằng một trong hai công cụ: Cygwin hoặc Windows Subsystem for Linux (WSL). Cả hai công cụ này đều tương thích với RQ.

Khởi tạo một tác vụ (Task)

Chúng ta sẽ bắt đầu bằng một ví dụ khởi tạo tác vụ đơn giản trong RQ để bạn làm quen với nó. Một tác vụ chỉ là một hàm Python. Sau đây là một ví dụ về tác vụ và chúng ta ta sẽ đưa nó vào một module mới là app/tasks.py:

app/tasks.py: Ví dụ về tác vụ nền

import time

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

import time

def example(seconds):

print('Starting task')

for i in range(seconds):

print(i)

time.sleep(1)

print('Task completed')

Tác vụ này sẽ nhận một tham số là thời gian tính bằng giây, sau đó in ra một bộ đếm trong thời gian này.

Thực thi một Tiến trình phục vụ RQ

Sau khi đã có tác vụ, chúng ta có thể bắt đầu thực thi một tiến trình phục vụ bằng lệnh rq worker:

(myenv) $ rq worker myblog-tasks
15:28:24: Worker rq:worker:36409de68562471396cf93c4bdfbd657: started, version 1.2.0
15:28:24: *** Listening on myblog-tasks...
15:28:24: Cleaning registries for queue: myblog-tasks18:55:06

(myenv) $ rq worker myblog-tasks

15:28:24: Worker rq:worker:36409de68562471396cf93c4bdfbd657: started, version 1.2.0

15:28:24: *** Listening on myblog-tasks...

15:28:24: Cleaning registries for queue: myblog-tasks18:55:06

Tiến trình phục vụ này sẽ kết nối với Redis và đợi đến khi có bất kỳ một tác vụ nào được gởi đến nó qua hàng đợi có tên là myblog-tasks. Nếu muốn sử dụng nhiều tiến trình phục vụ cùng lúc để tăng khả năng xử lý, bạn chỉ cần thực hiện lệnh rq worker nhiều lần để tạo ra nhiều thực thể tương ứng và cùng kết nối đến một hàng đợi. Khi một tác vụ được gởi đến và xuất hiện trong hàng đợi, nó sẽ được xử lý bởi một tiến trình phục vụ còn trống. Trong một môi trường sản phẩm chính thức, số tiến trình phục vụ ít nhất cũng phải bằng với số lượng CPU trong mỗi máy chủ.

Thực thi một tác vụ

Bây giờ bạn hãy mở một cửa sổ lệnh (terminal) thứ hai và kích hoạt môi trường ảo. Chúng ta sẽ sử dụng một phiên làm việc để thực thi tác vụ example() nhờ tiến trình phục vụ:

>>> from redis import Redis
>>> import rq
>>> queue = rq.Queue('myblog-tasks', connection=Redis.from_url('redis://'))
>>> job = queue.enqueue('app.tasks.example', 30)
>>> job.get_id()
'ddf1685b-3fd1-4e9b-9e4a-e1cabc58586f'

>>> from redis import Redis

>>> import rq

>>> queue = rq.Queue('myblog-tasks', connection=Redis.from_url('redis://'))

>>> job = queue.enqueue('app.tasks.example', 30)

>>> job.get_id()

'ddf1685b-3fd1-4e9b-9e4a-e1cabc58586f'

Lớp Queue trong Redis sẽ đại diện cho hàng đợi tác vụ trong ứng dụng. Để khởi tạo một thực thể của lớp này, chúng ta cần các tham số là tên hàng đợi và một đối tượng kết nối Redis (được khởi tạo với URL mặc định trong trường hợp này). Nếu server Redis của bạn nằm trên một máy chủ khác hoặc dùng một cổng khác, bạn sẽ phải cung cấp URL tương ứng.

Phương thức enqueue() của hàng đợi sẽ thêm một tác vụ vào hàng đợi. Tham số đầu tiên của phương thức này là tên của tác vụ bạn muốn thực hiện và được truyền trực tiếp bằng một đối tượng hàm hay là một chuỗi tham chiếu. Trong phạm vi của bài viết này, việc sử dụng chuỗi tham chiếu tiện lợi hơn vì chúng ta không cần phải thực hiện việc tham chiếu hàm trong ứng dụng. Các tham số còn lại cho phương thức enqueue() sẽ được truyền vào hàm được thực thi bởi tiến trình phục vụ.

Ngay khi gọi enqueue(), bạn sẽ thấy một số hoạt động trên trong cửa sổ lệnh đầu tiên đang chạy tiến trình phục vụ của RQ. Bạn sẽ thấy hàm example() được thực thi và in ra bộ đếm mỗi giây. Cùng lúc đó, bạn vẫn có thể tiếp tục thực hiện các chỉ thị trong phiên làm việc với ứng dụng trong cửa sổ lệnh thứ hai. Trong ví dụ trên, chúng ta gọi phương thức job.get_id() để nhận được định danh đã được gán cho tác vụ. Để kiểm tra tác vụ đã hoàn tất hay chưa, bạn có thể thử sử dụng một biểu thức thú vị khác với đối tượng job như sau:

>>> job.is_finished
False

1 2	>>> job.is_finished False

Nếu bạn dùng giá trị 30 như trong ví dụ trên, hàm này sẽ được thực thi trong 30 giây. Sau thời khoảng này, biểu thức job.is_finished sẽ trở thành True, đơn giản nhưng cũng rất hiệu quả.

Sau khi tác vụ đã được thi hành và hoàn tất, tiến trình phục vụ sẽ trở lại trạng thái chờ cho các tác vụ tiếp theo trong hàng đợi. Vì vậy, bạn có thể lặp lại việc gọi phương thức enqueue() với các tham số khác nếu bạn muốn tìm hiểu về nó. Dữ liệu cho các tác vụ sẽ được lưu trong hàng đợi trong một khoảng thời gian nhất định (500 giây theo mặc định), nhưng sẽ được xóa bỏ sau đó. Điều này rất quan trọng, bạn cần nhớ rằng hàng đợi sẽ không lưu lại lịch sử của các tác vụ đã được thực hiện qua nó.

Báo cáo tiến độ thực hiện tác vụ

Tác vụ trong ví dụ trên quá đơn giản và không có tác dụng thực tế nào. Thường thì đối với các tác vụ có thời gian thực thi kéo dài, chúng ta muốn có một số thông tin về tiến độ thực thi để hiển thị cho user. RQ có hỗ trợ cho nhu cầu này với thuộc tính meta của đối tượng tác vụ. Dưới đây là phiên bản cập nhật của tác vụ example() để tạo báo cáo (report) về tiến độ thực thi:

app/tasks.py: Tiến độ thực thi của tác vụ nền

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')

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')

Đoạn mã trên sử dụng hàm get_current_job() của RQ để truy cập thực thể tác vụ, tương tự như thực thể được hàng đợi tác vụ trả về cho ứng dụng khi nó gởi tác vụ đến hàng đợi. Thuộc tính meta của đối tượng tác vụ là một từ điển (dictionary) được tác vụ sử dụng để ghi các dữ liệu tùy biến cần được gởi đến ứng dụng trong quá trình thực thi. Trong ví dụ này, chúng ta ghi một mẩu tin gọi là progress với dữ liệu là tỉ lệ phần trăm hoàn thành tác vụ. Mỗi khi tiến độ được cập nhật, chúng ta sẽ gọi hàm job.save_meta() để thông báo với RQ lưu dữ liệu này vào Redis, nhờ đó ứng dụng có thể truy cập được.

Về phần ứng dụng (tạm thời chúng ta giả lập và tương tác với mã ứng dụng qua cửa sổ lệnh Python), chúng ta có thể thực thi tác vụ này và theo dõi tiến độ như sau:

>>> job = queue.enqueue('app.tasks.example', 30)
>>> 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

>>> job = queue.enqueue('app.tasks.example', 30)

>>> 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

Như bạn có thể thấy ở trên, chúng ta có thể truy cập và đọc thông tin cần thiết từ thuộc tính meta. Phương thức refresh() cần được sử dụng để cập nhật thông tin từ Redis.

Tạo cơ sở dữ liệu cho các tác vụ

Ví dụ trên giúp chúng ta hiểu các khái niệm cơ bản của việc thực thi tác vụ và theo dõi tiến trình thực thi. Tuy nhiên, quá trình này trở nên phức tạp hơn đối với các ứng dụng Web. Lý do là vì khi một trong các tác vụ này được khởi động theo yêu cầu từ phần mềm khách (client), yêu cầu sẽ kết thúc tại một thời điểm nào đó, và khi đó, các thông tin liên quan đến yêu cầu (context data) sẽ không còn nữa. Nhưng bởi vì chúng ta cần biết mỗi tác vụ đang được thực thi xuất phát từ yêu cầu của user nào, chúng ta sẽ cần dùng đến cơ sở dữ liệu để lưu lại các thông tin cần thiết. Sau đây là mô hình dữ liệu cho các tác vụ:

app/models.py: Mô hình Task

...
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

...

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

Một điểm khác biệt đáng lưu ý giữa mô hình này và các mô hình dữ liệu chúng ta đã sử dụng trước đây là chúng ta sử dụng dữ liệu kiểu chuỗi (string) cho khóa chính id thay vì số nguyên (integer). Lý do là vì chúng ta sẽ không dùng khóa chính do cơ sở dữ liệu tạo ra mà sẽ dùng id cho tác vụ do RQ tạo ra.

Mô hình dữ liệu này sẽ chứa tên đầy đủ của tác vụ (fully qualified name) – cũng là tên chúng ta sẽ gởi đến cho RQ, đặc tả của tác vụ (description) để hiển thị cho user, quan hệ với user đã yêu cầu tác vụ và một trường luận lý (boolean) để chỉ ra tác vụ đã hoàn tất hay chưa. Trường complete giúp chúng ta phân biệt giữa các tác vụ đã kết thúc và các tác vụ đang được thực thi bởi vì các tác vụ đang được thực thi cần được xử lý đặc biệt để hiển thị cập nhật về tiến độ.

Phương thức get_rq_job() là một hàm trợ giúp để truy cập đến một thực thể Job của RQ với id tương ứng của tác vụ từ mô hình dữ liệu. Trong hàm trợ giúp này, chúng ta gọi hàm Job.fetch() để truy cập đến thực thể Job từ dữ liệu tương ứng của Redis. Sau đó, phương thức get_progress() sẽ gọi hàm trợ giúp này và trả về tỉ lệ hoàn thành của tác vụ. Phương thức này dựa trên một số giả định như sau:

Nếu id của tác vụ trong mô hình dữ liệu không tồn tại trong hàng đợi của RQ, điều này có nghĩa là tác vụ đã hoàn thành và dữ liệu trong hàng đợi đã quá hạng và đã được xóa khỏi hàng đợi. Vì vậy, trong trường hợp này, tỉ lệ hoàn thành của tác vụ là 100.
Ở chiều ngược lại, nếu một tác vụ có tồn tại nhưng lại không có dữ liệu tương ứng từ thuộc tính meta, chúng ta có thể kết luận rằng tác vụ đã được sắp đặt để được thực thi nhưng chưa được thi hành trong thời điểm hiện tại. Vì vậy, tỉ lệ hoàn thành sẽ là 0 trong tình huống này.

Như thường lệ, chúng ta cần cập nhật cơ sở dữ liệu với các thay đổi này:

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

1 2	(myenv) $ flask db migrate -m "tasks" (myenv) $ flask db upgrade

Chúng ta cũng cần đưa mô hình dữ liệu mới này vào ngữ cảnh lệnh của ứng dụng (shell context) để có thể sử dụng nó trong một phiên làm việc từ chế độ dòng lệnh mà không cần phải tham chiếu:

myblog.py: Thêm mô hình dữ liệu tác vụ và ngữ cảnh lệnh

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}

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}

Tích hợp RQ với ứng dụng Flask

Địa chỉ để kết nối với dịch vụ Redis cần được thêm vào cấu hình cho ứng dụng:

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

class Config(object):

...

REDIS_URL = os.environ.get('REDIS_URL') or 'redis://'

Cũng như trước đây, địa chỉ thật để kết nối với Redis sẽ được đưa vào một biến môi trường, và nếu biến môi trường này không tồn tại, chúng ta sẽ sử dụng địa chỉ kết nối mặc định với giả thuyết rằng Redis đang chạy trên cùng một máy chủ với cổng mặc định.

Redis và RQ sẽ được khởi tạo trong hàm tạo ứng dụng:

app/__init.py: Tích hợp 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('myblog-tasks', connection=app.redis)

    ...

...

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('myblog-tasks', connection=app.redis)

...

Trong đoạn mã trên, app.task_queue sẽ là hàng đợi để nhận các tác vụ được gởi đến. Việc kết nối giữa hàng đợi với ứng dụng theo cách này rất tiện lợi vì nó cho phép chúng ta truy cập hàng đợi từ bất kỳ nơi nào trong ứng dụng qua biến current_app.task_queue. Để có thể dễ dàng gởi các tác vụ đến hàng đợi hoặc kiểm tra tình trạng của chúng, chúng ta sẽ tạo thêm một vài hàm trợ giúp trong lớp User:

app/models.py: Các hàm trợ giúp trong mô hình dữ liệu người sử dụng

...

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()

...

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()

Phương thức launch_task() sẽ đảm nhiệm việc gởi một tác vụ đến hàng đợi RQ đồng thời đưa các dữ liệu có liên quan vào cơ sở dữ liệu. Tham số name là tên của hàm được định nghĩa trong app/tasks.py. Khi gởi một tác vụ đến hàng đợi, hàm sẽ ghép chuỗi app.tasks. và tên của tác vụ để tạo thành tên đầy đủ của tác vụ (fully qualified name). Tham số description là một mô tả về tác vụ để hiển thị cho người sử dụng. Đối với hàm xuất file cho các bài viết sẽ được thực hiện ở phần sau, chúng ta sẽ sử dụng tên export_post và mô tả tương ứng là Exporting posts….Các tham số còn lại là các tham số theo vị trí (positional) và tham số theo từ khóa (keywords) và sẽ được truyền vào tác vụ. Hàm này bắt đầu bằng việc gọi phương thức enqueue() để gởi tác vụ đến hàng đợi. Sau khi phương thức enqueue() hoàn thành, nó sẽ trả về một đối tượng tác vụ có chứa id đã được RQ gán cho tác vụ, nhờ đó chúng ta có thể tạo ra dữ liệu cho đối tượng Task tương ứng trong cơ sở dữ liệu.

Lưu ý rằng hàm launch_task() thêm một đối tượng tác vụ mới vào phiên làm việc nhưng không gọi lệnh commit để lưu dữ liệu tác vụ. Nhìn chung, cách tốt nhất để làm việc với một phiên làm việc của cơ sở dữ liệu là ở các hàm ở mức độ cao hơn để có thể kết hợp nhiều cập nhật từ các hàm ở mức độ thấp hơn vào một giao dịch (transaction) cơ sở dữ liệu duy nhất. Tuy nhiên, đây cũng không phải một quy định bắt buộc, và trong thực tế thì bạn sẽ thấy một ngoại lệ khi chúng ta thực hiện một lệnh lưu dữ liệu trong một hàm con ở phần sau của bài viết này.

Phương thức get_tasks_in_progress() sẽ trả về một danh sách đầy đủ các hàm tương ứng với một user đang ở trong hàng đợi. Trong phần sau, chúng ta sẽ sử dụng phương thức này để hiển thị các thông tin về các hàm đang được thực thi cho người dùng.

Và cuối cùng, phương thức get_task_in_progress() là một phiên bản đơn giản hơn để trả về một tác vụ được chỉ định. Chúng ta không cho phép user thực hiện đồng thời hai tác vụ cùng loại trở lên, vì vậy, trước khi chúng ta bắt đầu một tác vụ, chúng ta có thể sử dụng phương thức này để kiểm tra xem có tác vụ cùng loại nào đang chạy hay không.

Gởi email từ các tác vụ RQ

Như đã đề cập từ đầu bài viết, khi tác vụ xuất file hoàn tất trong chế độ nền, user sẽ nhận được một email với một file đính kèm có chứa toàn bộ các bài viết của họ theo định dạng JSON. Để làm được điều này, chúng ta cần mở rộng chức năng gởi email đã được thực hiện trong Phần 10 ở hai phương diện. Thứ nhất, chúng ta cần hỗ trợ cho chức năng đính kèm file (attachment) để có thể gởi kèm file JSON. Thứ hai, hàm send_email() luôn thực hiện việc gởi email bất đồng bộ bằng cách sử dụng một luồng ở chế độ nền. Vì chúng ta đã có các tác vụ xuất file thực hiện trong chế độ nền trong một luồng riêng, việc có thêm một luồng khác chỉ để gởi email không hợp lý. Do đó, chúng ta cần hỗ trợ cả hai quá trình gởi email đồng bộ và bất đồng bộ.

Vì Flask-Mail có hỗ trợ cho chức năng đính kèm file, chúng ta chỉ cần sửa lại hàm send_email() để nhận thêm một tham số mới cho file đính kèm và sau đó thiết lập chúng trong đối tượng Message. Và để chuyển đổi giữa việc gởi email trong chế độ nền hoặc trong chế độ nổi (foreground) của ứng dụng, chúng ta sẽ thêm tham số sync như sau:

app/email.py: Gởi email với file đính kèm

...

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()

...

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()

Phương thức attach() trong lớp Message sẽ nhận ba tham số liên quan đến file đính kèm: tên file, kiểu file và dữ liệu trong file. Tên file chỉ là một tên gọi để hiển thị cho người sử dụng và không cần phải là một file thật sự. Kiểu file định nghĩa file thuộc loại nào để các chương trình nhận email có thể xử lý thích đáng. Ví dụ như nếu bạn gởi một file có kiểu là image/png, chương trình nhận email sẽ biết đó là một file ảnh và hiển thị hình ảnh tương ứng cho người nhận. Bởi vì chúng ta sẽ xuất bài viết trong các file theo định dạng JSON, chúng ta sẽ dùng kiểu file là application/json. Tham số thứ ba và cũng là tham số cuối cùng sẽ là một chuỗi ký tự hay byte với nội dung là dữ liệu trong file.

Để đơn giản hóa, tham số attachments trong hàm send_email() sẽ là một danh sách các tuple (danh sách tĩnh – tương tự như list nhưng không thay đổi được sau khi khởi tạo). Mỗi tuple sẽ gồm có ba phần tử tương ứng với ba tham số của hàm attach(). Với cách thiết lập này, chúng ta sẽ truyền mỗi phần tử trong danh sách dưới dạng tuple cho hàm attach() thay thì ba tham số riêng rẽ. Trong Python, nếu bạn có một list hay là một tuple với các tham số bạn muốn truyền cho một hàm, bạn có thể dùng cú pháp func(*args) để trình dịch Python tự động tạo ra các tham số tương ứng thay vì dùng cú pháp truyền thống func(args[0], args[1], args[2]). Ví dụ như nếu bạn có danh sách args = [1, 'foo'], cách gọi hàm mới này sẽ truyền hai tham số như khi bạn gọi hàm với biểu thức func(1, 'foo'). Nếu không có ký tự *, hàm sẽ được gọi với một tham số đơn là danh sách được truyền vào.

Và để thực hiện việc gởi email đồng bộ, chúng ta cần gọi hàm mail.send(msg) khi giá trị của sync là True.

Các hàm trợ giúp

Tác vụ example() trong ví dụ trên chỉ là một hàm độc lập và đơn giản, nhưng tác vụ để xuất các bài viết ra file sẽ cần sử dụng đến các chức năng trong ứng dụng như là truy cập cơ sở dữ liệu và gởi email. Bởi vì tác vụ này sẽ được thực thi trong một tiến trình riêng, chúng ta cần khởi tạo Flask-SQLAlchemy và Flask-Mail, và đến lượt các tác vụ này sẽ cần sử dụng các thông tin cấu hình từ thực thể ứng dụng. Vì vậy chúng ta sẽ thêm một thực thể ứng dụng Flask và ngữ cảnh ứng dụng vào đầu module app/tasks.py:

app/tasks.py: Tạo ứng dụng và ngữ cảnh

from app import create_app

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

from app import create_app

app = create_app()

app.app_context().push()

Thực thể ứng dụng được tạo ra trong module này bởi vì đây là module duy nhất được các tiến trình phục vụ RQ sẽ tham chiếu đến. Khi bạn sử dụng lệnh flask, module myblog.py trong thư mục gốc sẽ tạo ra thực thể ứng dụng, nhưng các tiến trình phục vụ của RQ hoàn toàn không biết về điều này. Vì vậy, chúng sẽ cần tạo ra thực thể ứng dụng riêng cho các tác vụ cần đến nó. Bạn đã thấy phương thức app.app_context() trong một vài nơi khác nhau trước đây và sử dụng phương thức push() với ngữ cảnh ứng dụng sẽ làm cho ứng dụng trở thành thực thể ứng dụng hiện hành, đồng thời cho phép các thư viện mở rộng như là Flask-SQLAlchemy sử dụng current_app.config để nhận được các tham số cấu hình của chúng. Nếu không có ngữ cảnh ứng dụng, biểu thức current_app sẽ báo lỗi.

Chúng ta sẽ tìm giải pháp để có thể tạo ra báo cáo tiến độ khi tác vụ đang chạy. Ngoài việc truyền các thông tin tiến độ qua từ điển job.meta, chúng ta cũng muốn gởi các thông báo với các thông tin này đến cho trình duyệt để cập nhập và hiển thị cho người sử dụng bằng kỹ thuật AJAX. Chúng ta sẽ thực hiện điều này với cơ chế thông báo đã được xây dựng trong Phần 21. Các cập nhật sẽ được thực hiện theo cách tương tự như với ô thông báo về số tin nhắn chưa được đọc. Khi server bắt đẩu sử dụng một template, nó sẽ đưa thông tin tiến độ “tĩnh” nhận được từ job.meta vào đó. Nhưng khi trang Web đã được tải về trình duyệt của người sử dụng, hệ thống thông báo sẽ cập nhật tỉ lệ hoàn thành của các tác vụ. Và để làm việc với hệ thống thông báo, chúng ta cần làm nhiều hơn so với ví dụ trên. Đầu tiên, chúng ta sẽ tạo ra một hàm đóng gói (wrapper) cho việc cập nhật tiến độ:

app/tasks.py: Thiết lập tiến độ của tác vụ

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()

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()

Tác vụ xuất file có thể gọi hàm _set_task_progress() để ghi nhận tỉ lệ hoàn thành. Hàm này sẽ ghi tỉ lệ phần trăm vào từ điển job.meta và lưu vào Redis. Sau đó, nó sẽ tiến hành nạp tác vụ tương ứng từ cơ sở dữ liệu và phương thức add_notification() có sẵn để tạo ra một thông báo đến task.user cũng là user đã yêu cầu tác vụ. Thông báo này sẽ được đặt tên là task_progress và sẽ có dữ liệu là một từ điển bao gồm hai đối tượng: định danh tác vụ và tỉ lệ phần trăm. Chúng ta sẽ thêm mã JavaScrip để xử lý kiểu thông báo mới này trong phần sau.

Hàm sẽ kiểm tra tiến độ của tác vụ. Nếu tác vụ đã hoàn thành, nó sẽ cập nhật thuộc tính complete của đối tượng tác vụ trong cơ sở dữ liệu. Cuối cùng, lệnh commit được thực hiện để bảo đảm rằng các đối tượng tác vụ và thông báo cho người sử dụng vừa được tạo ra qua hàm add_notification() được lưu vào cơ sở dữ liệu. Chúng ta cần rất cẩn trọng với cách thiết kế tác vụ cha để tránh các thay đổi trong cơ sở dữ liệu bởi vì lệnh commit cũng sẽ lưu lại các thay đổi này nếu có.

Xây dựng tác vụ xuất file

Đến đây, chúng ta đã có tất cả các thành phần cần thiết để xây dựng hàm xuất file. Cấu trúc của hàm này như sau:

app/tasks.py: Cấu trúc tổng quát của hàm xuất file

def export_posts(user_id):
    try:
        # Đọc các bài viết của user từ cơ sở dữ liệu
        # Gởi email có các bài viết đến user
    except:
        # Xử lý lỗi

def export_posts(user_id):

try:

# Đọc các bài viết của user từ cơ sở dữ liệu

# Gởi email có các bài viết đến user

except:

# Xử lý lỗi

Tại sao chúng ta phải sử dụng khối try/except block như trên? Mã ứng dụng trong các hàm xử lý yêu cầu không bị ảnh hưởng bởi các lỗi phát sinh ngoài dự kiến vì Flask sẽ tự động phát hiện và xử lý các lỗi này, đồng thời ghi nhận các tình huống lỗi trong các file nhật ký mà chúng ta đã thiết lập cho ứng dụng. Tuy nhiên, hàm này sẽ được thực thi trong một tiến trình riêng do RQ điều khiển chứ không phải Flask. Vì vậy nếu có một lỗi ngoài dự kiến xảy ra, tác vụ sẽ bị hủy bỏ, RQ sẽ hiển thị thông báo lỗi trên cửa sổ lệnh và trở lại chế độ chờ để đợi tác vụ mới. Vì vậy, nếu bạn không kiểm tra các thông tin trạng thái của các tiến trình phục vụ trong RQ hoặc đưa chúng vào một hệ thống nhật ký, bạn sẽ không bao giờ biết về các lỗi đã xảy ra.

Chúng ta sẽ đi vào chi tiết trong ba phần được chú thích trong đoạn mã trên, bắt đầu với phần đơn giản nhất là phần xử lý lỗi nằm ở cuối cùng:

app/tasks.py: Xử lý lỗi khi xuất file

import sys
...

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

import sys

...

def export_posts(user_id):

try:

...

except:

_set_task_progress(100)

app.logger.error('Unhandled exception', exc_info=sys.exc_info())

Khi lỗi xảy ra, chúng ta sẽ đánh dấu tác vụ là đã hoàn thành bằng cách thiết lập tiến độ là 100%, và sau đó lưu lại các thông tin liên quan vào hệ thống nhật ký, bao gồm thông báo lỗi, strack trace, và các thông tin được trả về từ hàm sys.exc_info(). Trong đoạn mã này, với việc sử dụng nhật ký ứng dụng của Flask, chúng ta cũng thấy tác dụng của hệ thống nhật ký đã được xây dựng trước đây. Ví dụ như trong Phần 7, chúng ta đã thiết lập cấu hình để ứng dụng gởi các thông báo lỗi đến người quản trị qua email. Khi sử dụng app.logger, chúng ta cũng có thể áp dụng cách làm này cho các lỗi trong tác vụ xuất file.

Tiếp theo, chúng ta sẽ viết mã nguồn để xuất file. Về bản chất, công việc này thật ra là thực hiện một truy vấn đến cơ sở dữ liệu, duyệt kết quả truy vấn và đưa dữ liệu tương ứng vào một từ điển:

app/tasks.py: Đọc các bài viết của user từ cơ sở dữ liệu

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)

        # Gởi email có các bài viết đến user
    except:
        ...

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)

# Gởi email có các bài viết đến user

except:

...

Với mỗi bài viết, hàm trên sẽ tạo ra một từ điển mới với hai phần từ, văn bản trong bài viết và thời gian bài viết được đăng. Thòi gian đăng bài sẽ theo chuẩn ISO 8601. Đối tượng datetime trong Python mà chúng ta sử dụng không chứa thông tin về múi giờ (timezone), vì vậy chúng ta sẽ thêm ký tự ‘Z’ phía sau để chỉ ra rằng đây là thời gian theo UTC.

Phần mã tiếp theo phức tạp hơn vì chúng ta cần theo dõi tiến độ. Chúng ta sẽ dùng biến đếm i và thực hiện một truy vấn dữ liệu khác trước khi khởi tạo vòng lặp để lấy tổng số bài viết và gán cho biến total_posts. Trong mỗi lần lặp, chúng ta có thể cập nhật tiến độ hoàn thành của tác vụ với một số trong khoảng từ 0 đến 100 bằng cách sử dụng i kết hợp với total_posts.

Bên cạnh đó, chúng ta cũng gọi hàm time.sleep(5) trong mỗi lần lặp để kéo dài thời gian thực hiện tác vụ và giúp chúng ta có thể thấy được tiến độ rõ ràng hơn dù khi tác vụ này chỉ xuất ra một số ít bài viết.

Và sau đây là phần cuối cùng của hàm này để gởi email đến user với tất cả các thông tin trong biến data dưới dạng file đính kèm:

app/tasks.py: Gởi email có bài viết đến user

import json
from flask import render_template
from app.email import send_email

# ...

def export_posts(user_id):
    try:
        # ...

        send_email('[Myblog] 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:
        # ...

import json

from flask import render_template

from app.email import send_email

# ...

def export_posts(user_id):

try:

# ...

send_email('[Myblog] 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:

# ...

Ở đây, công việc của chúng ta chỉ đơn giản là gọi hàm send_email(). File đính kèm được định nghĩa là một tuple với ba phần tử sẽ được truyền cho phương thức attach() của đổi tượng Message trong thư viện Flask-Mail. Phần tử thứ ba của tuple này là nội dung của file đính kèm và được tạo ra bởi hàm json.dumps() của Python.

Chúng ta cũng sử dụng hai template mới trong đoạn mã trên cho nội dung của email theo định dạng văn bản thông thường và HTML. Sau đây là nội dung của các template email dạng văn bản:

app/templates/email/export_posts.txt: Template cho email xuất file the dạng văn bản thường.

Dear {{ user.username }},

Please find attached the archive of your posts that you requested.

Sincerely,

Dear {{ user.username }},

Please find attached the archive of your posts that you requested.

Sincerely,

Và template email dạng HTML:

app/templates/email/export_posts.html: Template cho email xuất file the dạng HTML.

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

<p>Dear {{ user.username }},</p>

<p>Please find attached the archive of your posts that you requested.</p>

<p>Sincerely,</p>

Chức năng xuất file trong ứng dụng

Chúng ta đã chuẩn bị xong tất cả các thành phần chính để hỗ trợ tác vụ xuất file. Công việc còn lại chỉ là kết nối chức năng này với ứng dụng để user có thể gởi yêu cầu xuất file và nhận được email có các bài viết của họ.

Sau đây là hàm xử lý mới export_posts cho chức năng này:

app/main/routes.py: Hàm xử lý chức năng xuất file.

@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))

@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))

Hàm sẽ kiểm tra xem user có tác vụ xuất file nào trong hàng đợi hay không, và nếu có nó sẽ hiển thị một thông điệp thay vì bắt đầu một tác vụ mới vì chúng ta muốn tránh tình trạng một user có thể thực hiện nhiều tác vụ xuất file cùng lúc. Chúng ta kiểm tra điều kiện này bằng cách sử dụng phương thức get_task_in_progress() đã được tạo ra ở phần trước.

Nếu user hiện không có tác vụ xuất file nào, hàm sẽ bắt đầu một tác vụ mới bằng cách gọi hàm launch_task(). Tham số đầu tiên cho hàm này là tên của hàm sẽ được gởi đến tiến trình phục vụ RQ và được bắt đầu với chuỗi app.tasks. Tham số thứ hai là một chuỗi mô tả về tác vụ sẽ được hiển thị với người sử dụng. Cả hai giá trị này sẽ được lưu vào đối tượng Task trong cơ sở dữ liệu. Tiếp theo, hàm sẽ định hướng user đến trang hồ sơ cá nhân và kết thúc.

Chúng ta cũng cần tạo ra một liên kết để user có thể yêu cầu xuất file. Nơi thích hợp nhất cho liên kết này là trong trang hồ sơ cá nhân, bên dưới liên kết “Edit your profile” và chỉ được hiển thị khi nào user xem chính trang hồ sơ cá nhân của họ:

app/templates/user.html: Liên kết xuất file trong trang hồ sơ cá nhân.

                ...
                <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 %}

...

<p>

</a>

</p>

{% if not current_user.get_task_in_progress('export_posts') %}

<p>

</a>

</p>

...

{% endif %}

Ngoài ra, liên kết này cũng không được hiển thị nếu một user đang thực hiện tác vụ xuất file. Đó là lý do chúng ta có sử dụng điều kiện if not current_user.get_task_in_progress('export_posts') trong template trên.

Tại thời điểm này, chức năng xuất file đã hoạt động nhưng không trình bày kết quả cho user. Nếu muốn thử nghiệm, bạn có thể khởi động ứng dụng và tiến trình phục vụ RQ như sau:

Khởi động Redis
Trong một cửa sổ lệnh, khởi động một hoặc nhiều tiến trình phục vụ RQ với lệnh rq worker myblog-tasks
Trong một cửa sổ lệnh khác, khởi động ứng dụng Flask với lệnh flask run (và đừng quên thiết lập biến môi trường FLASK_APP trước khi nhập lệnh này).

Thông báo tiến độ

Để hoàn thành chức năng này, chúng ta muốn thông báo cho người sử dụng khi một tác vụ nền đang được thi hành và tiến độ theo tỉ lệ phần trăm. Để phục vụ cho mục đích này, chúng ta sẽ sử dụng một thành phần của Bootstrap gọi là cảnh báo (alert). Thành phần cảnh báo trong Bootstrap là một khối chữ nhật theo chiều ngang của màn hình và hiển thị một số thông tin cho người dùng. Trước đây, chúng ta đã sử dụng các khối cảnh báo này với màu xanh dương để hiển thị các thông điệp cho người dùng. Trong phần này, chúng ta sẽ sử dụng khối cảnh báo với màu xanh lá cây để hiển thị tình trạng của tác vụ xuất file và đặt nó bên dưới thanh định hướng như hình dưới đây:

app/templates/base.html: Cảnh báo cho tiến độ xuất file trong template gốc.

...
{% 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 %}
...

...

{% block content %}

{% if current_user.is_authenticated %}

{% with tasks = current_user.get_tasks_in_progress() %}

{% if tasks %}

{% for task in tasks %}

<span id="{{ task.id }}-progress">{{ task.get_progress() }}</span>%

</div>

{% endfor %}

{% endif %}

{% endwith %}

{% endif %}

...

{% endblock %}

...

Phương pháp hiển thị cảnh báo cho tác vụ xuất file gần hoàn toàn giống như cách hiển thị các thông điệp dùng flash. Điều kiện ngoài cùng sẽ bỏ qua các mã bên trong nếu user không đăng nhập. Đối với các user đã đăng nhập, chúng ta sẽ lấy danh sách các tác vụ đang được thực thi bằng cách gọi hàm get_tasks_in_progress() mà chúng ta đã tạo trong phần trên. Trong phiên bản hiện tại, chúng ta chỉ có thể nhận được nhiều nhất là một tác vụ vì chúng ta không cho phép một user thực hiện hai tác vụ hoặc hơn cùng lúc. Nhưng trong tương lại, chúng ta có thể cần hỗ trợ cho các tác vụ khác. Vì vậy, chúng ta cần viết mã để có thể sử dụng được trong trường hợp đó mà không cần phải thay đổi.

Với mỗi tác vụ, chúng ta sẽ tạo ra một phần tử cảnh báo (alert) trong trang. Màu sắc của phần tử cảnh báo được quy định bởi thuộc tính CSS thứ hai là alert-success trong trường hợp này (trong trường hợp của các thông điệp bằng flash trước đây, chúng ta sử dụng màu mặc định là alert-info). Nếu muốn tìm hiểu chi tiết về cấu trúc HTML của các phần tử alert, bạn có thể tham khảo tài liệu trực tuyến của Bootstrap. Phần tử alert sẽ hiển thị văn bản là giá trị của trường description trong đối tượng Task và tỉ lệ hoàn thành công việc.

Tỉ lệ hoàn thành công việc sẽ được đặt trong một phần tử <span> với thuộc tính id kèm theo. Lý do chúng ta phải sử dụng thuộc tính id là vì chúng ta sẽ cần cập nhật tỉ lệ này bằng JavaScript khi nhận được thông báo từ server. Chúng ta sẽ thiết lập các id này bằng cách ghép id của tác vụ với từ khóa –progress ở cuối cùng. Khi trình duyệt nhận được một thông báo từ server, thông báo này sẽ chứa id của tác vụ. Nhờ đó, chúng ta có thể dễ dàng tìm ra phần tử <span> cần được cập nhật với cú pháp #<id-của-tác-vụ>-progress.

Nếu bạn thử chạy ứng dụng ở thời điểm này, bạn sẽ thấy môt thanh tiến trình “tĩnh” mỗi khi bạn truy cập vào một trang mới và không được cập nhật trừ khi bạn chuyển sang trang khác.

Để có thể cập nhật “động” tỉ lệ phần trăm hoàn thành của tác vụ được hiển thị bởi phần tử <span>, chúng ta sẽ viết một hàm trợ giúp nhỏ bằng JavaScript:

app/templates/base.html: Hàm trợ giúp để cập nhật động tiến trình của tác vụ

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

...

{% block scripts %}

...

...

function set_task_progress(task_id, progress) {

$('#' + task_id + '-progress').text(progress);

}

</script>

...

{% endblock %}

Hàm này sẽ nhận một id của tác vụ và một giá trị theo tỉ lệ phần trăm, sau đó dùng jQuery để định vị phần tử <span> cho tác vụ này và cập nhật tỉ lệ hoàn thành bằng giá trị mới này. Chúng ta không cần phải kiểm tra xem phần từ cần định vị có tồn tại trong trang hay không, bởi vì nếu không tìm được phần tử này, jQuery sẽ không làm gì cả.

Tại thời điểm này, trình duyệt đã nhận được thông báo về tiến độ thực hiện từ server bởi vì hàm _set_task_progress() trong module app/tasks.py gọi hàm add_notification() mỗi khi tiến độ được cập nhật. Nếu bạn cảm thấy không hiểu lý do tại sao các thông báo có thể được gởi đến trình duyệt mà chúng ta không cần phải làm gì cả, đó là vì chúng ta đã xây dựng kiến trúc cho hệ thống thông báo trong Phần 21 theo cách tổng quát để nó có thể làm việc được với các chức năng khác nhau. Tất cả các thông báo được tạo ra qua hàm add_notification() sẽ được gởi đến trình duyệt thông qua các lần truy vấn định kỳ (polling) của trình duyệt để kiểm tra xem có thông báo mới hay không.

Tuy nhiên mã JavaScript để xử lý các thông báo này chỉ nhận ra các thông báo có tên gọi unread_message_count và bỏ qua các thông báo còn lại. Vì vậy, chúng ta cần thay đổi hàm này để xử lý cả các thông báo với tên là task_progress bằng cách gọi hàm set_task_progress() vừa được định nghĩa ở trên. Sau đây là phiên bản cập nhật của vòng lặp để xử lý các thông báo trong JavaScript:

app/templates/base.html: Hàm xử lý thông báo.

                        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;
                        }

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;

}

Trong đoạn mã trên, chúng ta cần xử lý hai loại thông báo khác nhau. Vì vậy chúng ta sẽ thay thế biểu thức if để kiểm tra tên của thông báo có phải là unread_message_count bằng biểu thức switch với tên của các thông báo mà chúng ta cần hỗ trợ. Nếu bạn không sử dụng qua ngôn ngữ lập trình C và các ngôn ngữ tương tự thuộc họ này, bạn có thể chưa gặp qua phát biểu switch. Đây là một cú pháp tiện lợi để thay thế một chuỗi dài các phát biểu if/elseif. Với thay đổi này, chúng ta chỉ cần thêm các khối case mới nếu chúng ta cần hỗ trợ thêm các thông báo mới.

Nếu bạn còn nhớ, dữ liệu dược các tác vụ RQ đưa vào thông báo task_progress là một từ điển với hai phần tử: task_id và progress, và đây cũng chính là hai tham số chúng ta cần để gọi hàm set_task_progress().

Nếu bạn thực thi ứng dụng tại thời điểm này, tỉ lệ hoàn thành của tác vụ trong phần tử alert sẽ được cập nhật mỗi 10 giây – mỗi khi các thông báo được gởi đến trình duyệt.

Và cuối cùng, bởi vì chúng ta có thêm một số chuỗi văn bản mới vào ứng dụng trong phần này, chúng ta cần cập nhật các chuỗi phiên dịch văn bản, bắt đầu với việc sử dụng Flask-Babel để cập nhật các file phiên dịch và sau đó bổ sung các chuỗi văn bản được dịch sang các ngôn ngữ khác trong các file tương ứng:

(myenv) $ flask translate update

1	(myenv) $ flask translate update

Nếu bạn đang sử dụng phần dịch thuật tiếng Việt, bạn có thể tải file có sẵn từ GitHub trong thư mục app/translations/vi/LC_MESSAGES/message.po và đưa vào dự án của bạn.

Sau khi đã cập nhật các văn bản được phiên dịch, bạn có thể tiến hành chuyển ngữ:

(myenv) $ flask translate compile

1	(myenv) $ flask translate compile

Một số lưu ý khi triển khai ứng dụng

Để kết thúc phần này, chúng ta sẽ thảo luận một số vấn đề khi triển khai ứng dụng với các thay đổi vừa được thực hiện ở trên. Để hỗ trợ cho các tác vụ nền chúng ta đã thêm hai thành phần mới vào ứng dụng là phần mềm Redis và một hoặc nhiều các tiến trình phục vụ. Hiển nhiên là chúng ta cần phải đưa các thành phần này vào trong quá trình triển khai ứng dụng trên các nền tảng khác nhau. Vì vậy, chúng ta sẽ xem xét lại các phương pháp triển khai ứng dụng trong các phần trước và thực hiện các thay đổi cho phù hợp.

Triển khai ứng dụng trên máy chủ Linux

Nếu bạn đang sử dụng ứng dụng trên máy chủ Linux, việc thêm Redis vào ứng dụng chỉ đơn giản là cài đặt gói này trong hệ điều hành của bạn. Ví dụ như với Ubuntu, bạn cần thực hiện lệnh sudo apt-get install redis-server .

Để thực thi các tiến trình phục vụ RQ, bạn cần xem lại và làm theo các hướng dẫn trong mục “Thiết lập cấu hình cho Gunicorn và Supervisor” trong Phần 17 để tạo ra cấu hình Supervisor thứ hai và thực hiện lệnh rq worker myblog-tasks thay vì gunicorn với cấu hình này. Nếu bạn muốn thực thi nhiều hơn một tiến trình phục vụ (đây cũng là phương pháp được đề nghị trong môi trường sản phẩm), bạn có thể dùng chỉ thị numprocs của Supervisor để chỉ ra số thực thể của tiến trình bạn muốn thực thi đồng thời.

Triển khai ứng dụng trên Heroku

Để triển khai ứng dụng trên Heroku, bạn cần thêm một server Redis vào tài khoản của bạn với cách làm tương tự như cách chúng ta thêm cơ sở dữ liệu Postgres. Bạn có thể cài đặt lớp miễn phí của Redis với lệnh sau đây:

$ heroku addons:create heroku-redis:hobby-dev

1	$ heroku addons:create heroku-redis:hobby-dev

Địa chỉ URL cho dịch vụ Redis sẽ được lưu trong biến môi trường REDIS_URL trong môi trường Heroku của bạn. Và đây cũng là giá trị được thiết lập trong ứng dụng, vì vậy, chúng ta không cần cập nhật cấu hình cho biến môi trường này.

Gói miễn phí của Heroku cho phép một dyno cho Web (Nếu không quen với thuật ngữ “dyno”, bạn có thể xem lại Phần 18) và một dyno cho tiến trình phục vụ, vì vậy bạn có thể sử dụng một tiến trình phục vụ đơn rq trong ứng dụng mà không cần phải lo ngại về phí tổn. Để làm điều này, bạn cần khai báo tiến trình phục vụ trong một dòng mới trong procfile của bạn:

web: flask db upgrade; flask translate compile; gunicorn myblog:app
worker: rq worker -u $REDIS_URL myblog-tasks

1 2	web: flask db upgrade; flask translate compile; gunicorn myblog:app worker: rq worker -u $REDIS_URL myblog-tasks

Sau khi triển khai các thay đổi này trên Heroku, bạn có thể bắt đầu tiến trình phục vụ với lệnh sau:

$ heroku ps:scale worker=1

1	$ heroku ps:scale worker=1

Triển khai ứng dụng với Docker

Nếu bạn đang triển khai ứng dụng trong các container của Docker, bạn sẽ cần tạo ra một container cho Redis. Bạn có thể sử dụng một image chính thức của Redis từ hệ thống danh bạ của Docker cho mục đích này với lệnh sau:

$ docker run --name redis -d -p 6379:6379 redis:3-alpine

1	$ docker run --name redis -d -p 6379:6379 redis:3-alpine

Khi thực thi ứng dụng, bạn cần liên kết container của Redis và thiết lập biến môi trường REDIS_URL tương tự như cách chúng ta liên kết ứng dụng với container của MySQL. Sau đây là lệnh khởi động ứng dụng với liên kết đến Redis:

$ docker run --name myblog -d -p 8000:5000 --rm -e SECRET_KEY=my-secret-key \
    -e MAIL_SERVER=smtp.googlemail.com -e MAIL_PORT=587 -e MAIL_USE_TLS=true \
    -e MAIL_USERNAME=<your-gmail-username> -e MAIL_PASSWORD=<your-gmail-password> \
    --link mysql:dbserver --link redis:redis-server \
    -e DATABASE_URL=mysql+pymysql://myblog:<database-password>@dbserver/myblog \
    -e REDIS_URL=redis://redis-server:6379/0 \
    myblog:latest

$ docker run --name myblog -d -p 8000:5000 --rm -e SECRET_KEY=my-secret-key \

-e MAIL_SERVER=smtp.googlemail.com -e MAIL_PORT=587 -e MAIL_USE_TLS=true \

-e MAIL_USERNAME=<your-gmail-username> -e MAIL_PASSWORD=<your-gmail-password> \

--link mysql:dbserver --link redis:redis-server \

-e DATABASE_URL=mysql+pymysql://myblog:<database-password>@dbserver/myblog \

-e REDIS_URL=redis://redis-server:6379/0 \

myblog:latest

Và cuối cùng, chúng ta cần thực thi một hoặc nhiều container cho các tiến trình phục vụ RQ. Bởi vì các tiến trình này dựa trên mã nguồn giống như của ứng dụng, chúng ta có thể sử dụng cùng image chúng ta đã dùng cho ứng dụng nhưng thay đổi lệnh khởi động để khởi động các tiến trình thay vì ứng dụng. Sau đây là một ví dụ để khởi động một tiến trình phục vụ với lệnh docker run:

$ docker run --name rq-worker -d --rm -e SECRET_KEY=my-secret-key \
    -e MAIL_SERVER=smtp.googlemail.com -e MAIL_PORT=587 -e MAIL_USE_TLS=true \
    -e MAIL_USERNAME=<your-gmail-username> -e MAIL_PASSWORD=<your-gmail-password> \
    --link mysql:dbserver --link redis:redis-server \
    -e DATABASE_URL=mysql+pymysql://myblog:<database-password>@dbserver/myblog \
    -e REDIS_URL=redis://redis-server:6379/0 \
    --entrypoint venv/bin/rq \
    myblog:latest worker -u redis://redis-server:6379/0 myblog-tasks

$ docker run --name rq-worker -d --rm -e SECRET_KEY=my-secret-key \

-e MAIL_SERVER=smtp.googlemail.com -e MAIL_PORT=587 -e MAIL_USE_TLS=true \

-e MAIL_USERNAME=<your-gmail-username> -e MAIL_PASSWORD=<your-gmail-password> \

--link mysql:dbserver --link redis:redis-server \

-e DATABASE_URL=mysql+pymysql://myblog:<database-password>@dbserver/myblog \

-e REDIS_URL=redis://redis-server:6379/0 \

--entrypoint venv/bin/rq \

myblog:latest worker -u redis://redis-server:6379/0 myblog-tasks

Việc thay đổi lệnh khởi động của một image Docker tương đối rắc rối vì lệnh này có hai phần khác nhau. Tham số --entrypoint chỉ nhận vào một tên file thực thi, nhưng các tham số kèm theo (nếu có) cần được đưa vào sau tên image và thẻ tại cuối dòng lệnh. Cũng cần lưu ý rằng chúng ta phải sử dụng đường dẫn thích hợp cho rq là myenv/bin/rq để nó có thể hoạt động mà không cần kích hoạt môi trường ảo.

Chúng ta sẽ tạm ngừng ở đây. Hẹn gặp bạn trong phần tiếp theo.

Hướng dẫn lập trình Flask – Phần 22: Tìm hiểu về tác vụ nền

Giới thiệu về Hàng đợi tác vụ (Task Queue)

Sử dụng RQ

Khởi tạo một tác vụ (Task)

Thực thi một Tiến trình phục vụ RQ

Thực thi một tác vụ

Báo cáo tiến độ thực hiện tác vụ

Tạo cơ sở dữ liệu cho các tác vụ

Tích hợp RQ với ứng dụng Flask

Gởi email từ các tác vụ RQ

Các hàm trợ giúp

Xây dựng tác vụ xuất file

Chức năng xuất file trong ứng dụng

Thông báo tiến độ

Một số lưu ý khi triển khai ứng dụng

Triển khai ứng dụng trên máy chủ Linux

Triển khai ứng dụng trên Heroku

Triển khai ứng dụng với Docker

Leave a Reply Cancel reply