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

import logging
from typing import List

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.template.loader import render_to_string

from ferrosoft.apps.dataimport.models import LogEntry, LogHighlight
from ferrosoft.apps.ferrobase.consumers import EntityIdRequired
from ferrosoft.apps.ferrobase.frontend import FrontendUIUpdater




class ImportJobUIUpdater(FrontendUIUpdater):
    """
    Allow server-side code to push updates regarding ImportJob (LogEntry)
    to the browser.
    """

    def __init__(self, entity_id: str = None):
        super().__init__()
        if entity_id is None:
            raise EntityIdRequired
        self._group_name = "import_job_%s" % entity_id



    def send_log_entries(self, entries: List[LogEntry]):
        channel_layer = get_channel_layer()
        data = {
            "type": "pass_on",
            "state": "log_entry_created",
            "includes_error": LogHighlight.DANGER
            in map(lambda entry: entry.highlight, entries),
            "content": render_to_string(
                "dataimport/job/log-entry.html",
                context={
                    "log_entries": reversed(entries),
                },
            ),
        }
        async_to_sync(channel_layer.group_send)(self._group_name, data)

        for entry in entries:
            logging.info(str(entry))




    def send_log_entry(self, job, message, line_number=None, highlight=None):
        self.send_log_entries(
            [LogEntry.create(job, line_number or 0, message, highlight)]
        )




    def send_import_finished(self, error: bool, line_count: int):
        channel_layer = get_channel_layer()
        data = {
            "type": "pass_on",
            "state": "import_finished",
            "error": error,
            "lineCount": line_count,
        }
        async_to_sync(channel_layer.group_send)(self._group_name, data)