Source code for ferrosoft.apps.dataimport.websocket.consumer

from django.template.loader import render_to_string

from ferrosoft.apps.dataimport.websocket.updater import ImportJobUIUpdater
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.consumers import UIEventsConsumer
from ferrosoft.apps.ferrobase.services.taskqueue import background_task_tenant




class ImportJobUIEventsConsumer(UIEventsConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(ImportJobUIUpdater, *args, **kwargs)



    def receive_json_tenant_context(self, content: dict):
        match content.get("action", ""):
            case "start_import":
                self._start_import()




    def pass_on(self, event):
        self.send_json(event)


    def _start_import(self):
        background_task_tenant(
            "ferrosoft.apps.dataimport.queue.process_import_job",
            self.entity_id,
        )
        self.send_json(
            {
                "state": "import_started",
                "content": render_to_string(
                    "dataimport/job/import-started.html",
                    context={
                        "now": now_utc(),
                    },
                ),
            }
        )