Xây dựng dựng dụng REST API đơn giản với FLASK - Part 3

Halu mọi người, mình đã quay trở lại. Như đã hứa ở bài viết trước, hôm nay mình sẽ hướng dẫn các bạn cách làm việc với redis trên ứng dụng flask của các bạn. VD này, mình chỉ dùng redis để chạy một function nho nhỏ thôi nha. để các bạn tập trung vào việc sử dụng redis như thế nào ạ. Nên nếu có gì các bạn đừng ném đá là được. nhiều gạch đá quá sợ xây vài cái biệt thự cũng không hết. 😦

Install requirement

  • Redis (3.0.7)
  • Python Redis (2.10.15)
  • RQ (0.5.6)

À để đảm bảm redis có thể hoạt động thì bạn nên kiểm tra, và chắc chắn máy tính của bạn đã được cài đặt thành công Redis nha. Hoặc nếu chưa, bạn có thể download và cài đặt tại đây.

Khi máy của bạn đã được cài đặt thành công, hãy khởi động redis server:

$ redis-server

Tiếp theo cài đặt python redis

$ cd flask_example
$ pip install redis==2.10.5 rq==0.5.6
$ pip freeze > requirements.txt

OK vậy là bước cài đặt các thư viện đã xong. Việc tiếp theo chúng ta cần thực hiện là initial worker để redis có thể hoạt động được trên project của mình.

Set up Worker

Ở bước này, bạn cần phải tạo một file mới, tên worker.py và thêm vào đoạn code sau:

import os

import redis
from rq import Worker, Queue, Connection

listen = ['default']

redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue, listen)))
        worker.work()

Ở đây, chúng ta sẽ connect tới redis server tại localhost:6379, bây giờ hãy kick hoạt worker nào

$ cd flask_example
$ python worker.py
13:27:33 RQ worker 'rq:worker:framgia0279-pc.10290' started, version 0.5.6
13:27:33 
13:27:33 *** Listening on default...

Vậy là phần work đã có thể hoạt động trên project của bạn. Tiếp theo chúng ta cần update lại app.py để đưa conjob vào hàng đợi (queue).

Update app.py

Thêm các đoạn import sau vào file app.py

from rq import Queue
from rq.job import Job
from worker import conn

Tại file config.py bạn thêm đoạn config: q = Queue(connection=conn) tại bước này file config của bạn sẽ trông giống như vậy:

app = Flask(__name__)
app.config.from_object(os.environ['APP_SETTINGS'])
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)

q = Queue(connection=conn)

from models import *

Đoạn code trên được dùng để thiết lập kết nối Redis và khởi tạo một hàng đợi dựa trên kết nối đó.

Việc kết nối Redis coi như đã hoàn thành. Chúng ta còn thiếu method để thực hiện hàng đợi. Tôi ví dụ một function đơn giản để các bạn có thể hình dung nhé. VD là việc lấy thời gian hiện tại chẳng hạn. Bạn tạo một function như sau:

from datetime import datetime, timezone
import pytz

def get_time_example(timezone):
    tz = pytz.timezone(timezone)
    time = datetime.now(tz)

    return time
    
@app.route('/get_time', methods=['GET', 'POST'])
def get_time():
    if request.method == 'POST':
        timezone = request.form['timezone']
        
        job = q.enqueue_call(
            func=get_time_example, args=(timezone,), result_ttl=180
        )
        print(job.get_id())

Bạn hoàn toàn có thể thay thế function get_time_example() bằng 1 function khác để thực hiện job của mình. Ở ví dụ này, function get_time_example() của tôi chấp nhận 1 giá trị truyền vào là timezone.

Bạn có thể dễ dàng nhìn thấy tại /get_time route đoạn code:

job = q.enqueue_call(
    func=get_time_example, args=(timezone,), result_ttl=1800
)
print(job.get_id())

Ở đây chúng ta đã sử dụng hàng đợi mà chúng ta đã khởi tạo trước đó và gọi hàm enqueue_call(). Việc làm này đã thêm một job mới vào hàng đợi và công việc đó là chạy hàm get_time_example() với tham số timezone.

Tham số result_ttl = 1800 cho RQ biết thời gian giữ kết quả của job là trong khoảng 1800 giây. Sau đó, chúng tôi xuất ra id của job vừa thực hiện. Bạn có thể sử dụng ID này để kiểm tra job đã đã hoàn thành hay chưa.

Nào. hãy thử check xem kết quả chạy job vừa rồi đã hoàn thành chưa nhé.

@app.route("/results/<job_key>", methods=['GET'])
def get_results(job_key):

    job = Job.fetch(job_key, connection=conn)

    if job.is_finished:
        return str(job.result), 200
    else:
        return "Nay!", 202

Wow, vậy là xong rồi. Giờ bạn có thể chạy ứng dựng của mình và trải nghiệm kết quả nhé. Hy vọng với chút kiến thức ít ỏi tôi tìm hiểu được và chia sẻ lại có thể giúp các bạn dễ dàng áp dụng được redis vào ứng dụng của chính mình.

Tài liệu tham khảo:

https://realpython.com/flask-by-example-implementing-a-redis-task-queue/