# SPDX-FileCopyrightText: Copyright (C) 2026 Omid Jafari <omidjafari.com>
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from pathlib import Path
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from nlpmed_portal.annotations import task_dispatch
from nlpmed_portal.annotations.api.serializers import ImportJobSerializer
from nlpmed_portal.annotations.api.viewsets.base import BaseViewSet
from nlpmed_portal.annotations.models import ImportJob
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.permissions import has_permission
from nlpmed_portal.annotations.utils import get_required_fields_for_import
from nlpmed_portal.annotations.utils import read_dataframe
[docs]
class ImportJobViewSet(BaseViewSet):
serializer_class = ImportJobSerializer
queryset = ImportJob.objects.select_related("user", "project").all()
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_importjob",
"HEAD": "view_importjob",
"OPTIONS": "view_importjob",
"POST": "add_importjob",
"PUT": "change_importjob",
"PATCH": "change_importjob",
"DELETE": "delete_importjob",
}
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
queryset = super().get_queryset()
project_id = self.request.query_params.get("project_id")
if project_id:
if not project_id.isdigit():
raise ValidationError(
{"project_id": "Invalid project_id. Must be an integer."},
)
queryset = queryset.filter(project_id=project_id)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="project",
)
[docs]
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
job: ImportJob = serializer.save(user=request.user)
import_target_perm = f"import_{job.import_target}"
if not has_permission(
user=request.user,
codename=import_target_perm,
project=job.project,
):
return Response(
{"error": f"No permission to import {job.import_target}"},
status=status.HTTP_403_FORBIDDEN,
)
try:
df_preview = read_dataframe(
Path(job.import_file.path),
job.import_format,
nrows=5,
)
preview_data = {
"columns": list(df_preview.columns),
"rows": df_preview.values.tolist(),
}
except Exception as exc: # ruff: ignore[blind-except]
job.job_status = "failed"
job.error_message = str(exc)
job.save()
job.delete_import_file()
return Response(
{"error": f"Failed to parse preview: {exc}"},
status=status.HTTP_400_BAD_REQUEST,
)
job_data = self.get_serializer(job).data
job_data["preview_data"] = preview_data
required_fields = get_required_fields_for_import(
import_target=job.import_target,
import_missing_patients=job.import_missing_patients,
)
if not required_fields:
job.job_status = "failed"
job.error_message = f"Unsupported import target: {job.import_target}"
job.save()
job.delete_import_file()
return Response(
{"error": f"Invalid import_target={job.import_target}"},
status=status.HTTP_400_BAD_REQUEST,
)
job_data["required_fields"] = required_fields
return Response(job_data, status=status.HTTP_201_CREATED)
[docs]
@action(detail=True, methods=["POST"])
def confirm(self, request, pk=None):
job: ImportJob = self.get_object()
import_target_perm = f"import_{job.import_target}"
if not has_permission(
user=request.user,
codename=import_target_perm,
project=job.project,
):
return Response(
{"error": f"No permission to import {job.import_target}"},
status=status.HTTP_403_FORBIDDEN,
)
if job.job_status != "uploaded":
return Response(
{"error": f"Cannot confirm import in status {job.job_status}."},
status=status.HTTP_400_BAD_REQUEST,
)
job.column_map = request.data.get("column_map", {})
job.date_format_map = request.data.get("date_format_map", {})
job.job_status = "in_progress"
job.save()
task_dispatch.dispatch_import_job(job.id)
return Response({"detail": "Import job started."}, status=status.HTTP_200_OK)
[docs]
@action(detail=True, methods=["GET"])
def status(self, request, pk=None):
job: ImportJob = self.get_object()
data = self.get_serializer(job).data
return Response(data, status=status.HTTP_200_OK)