Nhân viên cần cẩu Django để gửi trạng thái thời gian thực và thông báo kết quả đến giao diện người dùng

Question

Trong ứng dụng django tôi đang chạy tác vụ không đồng bộ và muốn hiển thị tiến trình, lỗi, v.v ... cho người dùng. Nếu có lỗi, người dùng phải được chuyển hướng đến trang có thêm đầu vào hoặc một số hành động cần thiết để khắc phục sự cố. Cách tốt nhất để giao tiếp từ công việc cần tây trở lại giao diện người dùng là gì?

Dưới đây là một cấu trúc cơ bản trong mã giả:

# views.py 
from tasks import run_task 

def view_task(): 
    run_task.delay() 
    return render(request, 'template.html') 

# tasks.py 
from compute_module import compute_fct 

@shared_task 
def run_task(): 
    result = compute_fct() 

    # how to catch status update messages from compute_module while compute_fct is running?? 

    if result == 'error': 
     handle_error() 
    else: 
     handle_succes()  

# compute_module 
import pandas as pd 

def compute_fct(): 
    # send message: status = loading file 
    df = pd.read_csv('test.csv') 
    # send message: status = computing 
    val = df['col'].mean() 

    if val is None: 
     return {'status':'error'} 
    else: 
     return {'status':'success','val':val} 

Những gì tôi tưởng muốn:

• compute_module.py Module này sử dụng python logger bản địa. Bằng cách tách các nhiệm vụ, tôi muốn giữ cho việc ghi nhật ký càng chung càng tốt và sử dụng các trình ghi chuẩn python/django. Nhưng chúng dường như không được thiết kế để gửi tin nhắn đến giao diện người dùng.
• cần tây nhiệm vụ bằng cách nào đó xử lý các bản ghi và thay vì hiển thị chúng trên stdout chuyển hướng họ Pusher
• chương trình front-end js và xử lý các thông điệp

Có thể có cách tiêu chuẩn giao tiếp giữa người lao động cần tây và phía trước kết thúc mà tôi không biết. kịch bản này phải xảy ra thường xuyên và tôi ngạc nhiên vì nó rất khó thực hiện. theo cách mà hàng đợi thông báo thỏmq hoặc sns aws nên được thiết kế cho việc này. dưới đây là những tài nguyên mà tôi đã xem xét nhưng không cảm thấy một trong số chúng hoạt động rất tốt nhưng có lẽ tôi chỉ đang bối rối.

đăng nhập: đây có vẻ là thêm về cách đăng nhập vào phía máy chủ, không gửi tin nhắn đến người dùng

• http://docs.celeryproject.org/en/latest/userguide/tasks.html#logging
• https://docs.djangoproject.com/en/2.0/topics/logging/
• http://oddbird.net/2017/04/17/async-notifications/
• https://www.google.com/search?q=celery+worker+send+message+to+front+end

Cần tây cam có vẻ là về nhiệm vụ giám sát quản trị, không phải là sen thông điệp ding để sử dụng

• http://docs.celeryproject.org/en/latest/userguide/monitoring.html

đẩy tôi thích nhưng tôi không muốn có compute_module.py đối phó với nó. Đó là Ví dụ tôi không muốn làm bất kỳ sự tích hợp pusher.com nào bên trong compute_module.py. Đoán tôi có thể vượt qua một đối tượng pusher mà đã được khởi tạo nên các mô-đun chỉ có thể đẩy thông điệp nhưng một lần nữa tôi muốn nó được generic

• https://blog.pusher.com/improve-user-experience-app-real-time-progress-bar-tutorial/
• https://blog.pusher.com/django-pusherable/

Answer 1

Cách duy nhất tôi đã quản lý để có được trạng thái thời gian thực là chỉ cần đặt một số SQL viết/api cuộc gọi vào nhiệm vụ chính nó. Làm việc với giá trị trả về của tác vụ dễ dàng hơn nhiều vì bạn chỉ có thể viết một lớp nhiệm vụ tùy chỉnh.

Tôi không hoàn toàn chắc chắn cách hoạt động này bằng cách sử dụng Django nhưng nó sẽ giống như thế này.

class CustomTask(celery.Task): 
    def __call__(self, *args, **kwargs): 
     self.start_time = time.time() 

    def on_success(self, retval, task_id, args, kwargs): 
     do_success_stuff() 

    def on_failure(self, exc, task_id, args, kwargs, einfo): 
     do_failure_stuff() 

@shared_task(base=CustomTask) 
def do_stuff(): 
    return create_widgets() 

Danh sách đầy đủ có thể được tìm thấy ở đây: http://docs.celeryproject.org/en/latest/userguide/tasks.html#handlers

Answer 2

Ok như vậy dưới đây là mã giả cho làm thế nào tôi đã giải quyết nó cho bây giờ. Về cơ bản, tôi sử dụng https://pusher.com/docs/javascript_quick_start và phía máy chủ chuyển đối tượng được tạo thành vào compute_module. Một nhược điểm là các thông điệp pusher là ephermeral vì vậy tôi sẽ phải làm một số công việc phụ trong LogPusher để lưu trữ chúng trong một db, một cái gì đó cho một ngày ...

Ngoài ra trong thực hiện thực tế của tôi, tôi kích hoạt nhiệm vụ thông qua một cuộc gọi $.post() ajax trong $(document).ready() vì các tác vụ nhỏ hoàn thành quá nhanh nên người dùng sẽ không bao giờ thấy thông báo đẩy vì kết nối không được thiết lập (quay lại vấn đề thông báo lịch sử đó).

Một tuyến đường thay thế mà tôi đã không được đề cập ở trên là https://channels.readthedocs.io/en/latest/

# views.py 
from tasks import run_task 

def view_task(): 
    run_task.delay('event') 
    return render(request, 'template.html', 'pusher_event':'event') 


# tasks.py 
import pusher 
from django.conf import settings 
from compute_module import compute_fct 

class LogPusher(object): 
    def __init__(self, event): 
     self.pusher_client = pusher.Pusher(app_id=settings.PUSHER_APP_ID, 
         key=settings.PUSHER_KEY, 
         secret=settings.PUSHER_SECRET, 
         cluster=settings.PUSHER_CLUSTER, ssl=True) 
     self.event = event 

    def send(self, data): 
     self.pusher_client.trigger(settings.PUSHER_CHANNEL, self.event, json.dumps(data)) 

@shared_task 
def run_task(pusher_event): 

    log_pusher = LogPusher(pusher_event) 
    result = compute_fct(log_pusher) 

    # how to catch status update messages from compute_module while compute_fct is running?? 

    if result == 'error': 
      log_pusher.send('status':'error') 
    else: 
      log_pusher.send('status':'success') 


# compute_module.py 
import pandas as pd 

def compute_fct(log_pusher): 
    # send message: status = loading file 
    log_pusher.send('status':'loading file') 
    df = pd.read_csv('test.csv') 
    # send message: status = computing 
    log_pusher.send('status':'computing') 
    val = df['col'].mean() 

    if val is None: 
     return {'status':'error'} 
    else: 
     return {'status':'success','val':val} 


# context_processors.py 
# see https://stackoverflow.com/questions/433162/can-i-access-constants-in-settings-py-from-templates-in-django 
from django.conf import settings 

def pusher(request): 
    return {'PUSHER_KEY': settings.PUSHER_KEY, 'PUSHER_CLUSTER': settings.PUSHER_CLUSTER , 'PUSHER_CHANNEL': settings.PUSHER_CHANNEL } 


# template.html 
<script> 

var pusher = new Pusher("{{PUSHER_KEY}}", { 
    cluster: "{{PUSHER_CLUSTER}}", 
    encrypted: true  
}); 

var channel = pusher.subscribe("{{PUSHER_CHANNEL}}"); 
channel.bind("{{pusher_event}}", function(data) { 
    // process data 
}); 

</script>