import logging
import traceback
from dataclasses import dataclass
from typing import Any
from asgiref.sync import sync_to_async
from django.utils.translation import gettext as _
from ferrosoft.apps.dataimport import convert
from ferrosoft.apps.dataimport.models import (
ImportJob,
MappingCollection,
FieldMapping,
LogHighlight,
ImportStatus,
LogEntry,
)
from ferrosoft.apps.dataimport.websocket.updater import ImportJobUIUpdater
from ferrosoft.apps.ferrobase.services.taskqueue import (
tenant_task,
wait_for_success,
)
from ferrosoft.importutils import import_object, ForbiddenModule
[docs]
class ProcessorError(Exception):
pass
[docs]
class ForbiddenProcessor(ProcessorError):
pass
[docs]
class ProcessorNotFound(ProcessorError):
pass
[docs]
@dataclass
class ProcessorArgs:
job: ImportJob
ui_updater: ImportJobUIUpdater
[docs]
@dataclass
class ModelProcessorArgs:
job: ImportJob
ui_updater: ImportJobUIUpdater
model: object
processor_state: Any
line_number: int
def _run_processor(qualified_name: str, **processor_kwargs):
"""Run processor function.
:param str qualified_name: Fully qualified function name starting with ferrosoft.apps.
:param processor_kwargs: Keyword arguments to pass to processor.
:raises ProcessorNotFound: if function does not exist.
:raises ForbiddenProcessor: if function name does not start with ferrosoft.apps.
:raises ProcessorError: if the function raised an exception.
:return: return value from processor function.
"""
try:
func = import_object(qualified_name)
except ForbiddenModule:
raise ForbiddenProcessor(qualified_name)
if func is None:
raise ProcessorNotFound(qualified_name)
try:
return func(**processor_kwargs)
except Exception as e:
raise ProcessorError("Processor %s failed" % qualified_name) from e
def _run_processor_guarded(
processor_name: str, job: ImportJob, updater: ImportJobUIUpdater
):
try:
return _run_processor(processor_name, pargs=ProcessorArgs(job, updater))
except ProcessorError as e:
logging.error("Failed running %s: %s", processor_name, str(e))
logging.error(traceback.format_exc())
updater.send_log_entries(
[
LogEntry.create(
job,
0,
_("Failed running data processor"),
LogHighlight.DANGER,
),
]
)
[docs]
@sync_to_async
@tenant_task()
def process_import_job(import_job_id):
job = ImportJob.objects.get(pk=import_job_id)
# Updater used to notify browser via Websocket.
updater = ImportJobUIUpdater(import_job_id)
# Result of processor before job, passed along to other processors.
processor_state = None
if job.processor_before_job:
processor_state = _run_processor_guarded(job.processor_before_job, job, updater)
try:
# Immediately set processing status to prevent starting the same job several times while it is being processed.
job.set_status(ImportStatus.PROCESSING)
convert.import_job_lines(job, processor_state)
updater.send_log_entries(
[
LogEntry.create(
job,
0,
_("All lines have been enqueued for processing"),
highlight=LogHighlight.INFO,
),
]
)
except Exception as e:
logging.error("failed processing import job: %s", str(e))
job.set_status(ImportStatus.ERRORED)
updater.send_import_finished(True, job.line_count)
else:
logging.info(
"Waiting for data import to complete with %d lines", job.line_count
)
# Job has been updated with line count. Wait for as many tasks to finish in the queue group.
errored = not wait_for_success(result_group=job.queue_group_name(), count=job.line_count)
if job.processor_after_job:
_run_processor_guarded(job.processor_after_job, job, updater)
job.set_status(ImportStatus.ERRORED if errored else ImportStatus.FINISHED)
updater.send_import_finished(errored, job.line_count)
if errored:
logging.error("Job finished with errors: %s", job.pk)
else:
logging.info("Job finished successfully: %s", job.pk)
[docs]
@sync_to_async
@tenant_task()
def process_import_line(job_data, line_no, line, processor_state) -> bool:
try:
if not isinstance(job_data, dict):
logging.error("job_data must be a dict")
return False
if not isinstance(line, dict):
logging.error("line must be a dict")
return False
updater = ImportJobUIUpdater(job_data.get("id", None))
# Create in-memory objects to avoid hitting the database and producing SELECT storms
# due to this function running many times in parallel.
collection_data = job_data.pop("collection", {})
field_mappings = [
FieldMapping(**data) for data in collection_data.pop("field_mappings", [])
]
collection = MappingCollection(**collection_data)
job = ImportJob(**job_data)
model, log_entries = convert.convert_line(
line_no, line, job, field_mappings, collection
)
if len(log_entries) == 0:
logging.error("convert_line did not return any logs")
return False
else:
updater.send_log_entries(log_entries)
if log_entries[-1].highlight != LogHighlight.SUCCESS:
return False
if model is not None and job.model_processor:
_run_processor(
job.model_processor,
pargs=ModelProcessorArgs(job, updater, model, processor_state, line_no),
)
except ProcessorError as e:
logging.error("failed running model processor: %(error)s", {"error": str(e)})
logging.error(traceback.format_exc())
return False
except Exception as e:
logging.error("failed processing import line: %(error)s", {"error": str(e)})
logging.error(traceback.format_exc())
return False
return True