feat: add import taxonomy endpoint

openedx_tagging/core/tagging/import_export/api.py

@@ -48,12 +48,32 @@
 
 from django.utils.translation import gettext as _
 
+from .. import api as taxonomy_api
 from ..models import TagImportTask, TagImportTaskState, Taxonomy
 from .exceptions import TagImportError
 from .import_plan import TagImportPlan, TagImportTask
 from .parsers import ParserFormat, get_parser
 
 
+def create_taxonomy_and_import_tags(
+    taxonomy_name: str,
+    taxonomy_description: str,
+    file: BytesIO,
+    parser_format: ParserFormat
+) -> bool:
+    """
+    Create a taxonomy and import the tags from `file`
+    """
+    taxonomy = taxonomy_api.create_taxonomy(taxonomy_name, taxonomy_description)
+
+    import_success = import_tags(taxonomy, file, parser_format)
+
+    if not import_success:
+        taxonomy.delete()
+
+    return import_success
+
+
 def import_tags(
     taxonomy: Taxonomy,
     file: BytesIO,

openedx_tagging/core/tagging/rest_api/v1/serializers.py

@@ -7,6 +7,7 @@
 from rest_framework.reverse import reverse
 
 from openedx_tagging.core.tagging.data import TagData
+from openedx_tagging.core.tagging.import_export.parsers import ParserFormat
 from openedx_tagging.core.tagging.models import ObjectTag, Tag, Taxonomy
 
 
@@ -175,3 +176,43 @@ class TaxonomyTagDeleteBodySerializer(serializers.Serializer):  # pylint: disabl
         child=serializers.CharField(), required=True
     )
     with_subtags = serializers.BooleanField(required=False)
+
+
+class TaxonomyImportBodySerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """
+    Serializer of the body for the Taxonomy Import action
+    """
+    taxonomy_name = serializers.CharField(required=True)
+    taxonomy_description = serializers.CharField(default="")
+    file = serializers.FileField(required=True)
+
+    def validate_file(self, file):
+        """
+        Validates the file extension
+        """
+        filename = file.name
+        ext = filename.split('.')[-1]
+        parser_format = getattr(ParserFormat, ext.upper(), None)
+        if not parser_format:
+            raise serializers.ValidationError(f'File type not supported: {ext.lower()}')
+
+        return file
+
+    def get_parser_format(self, obj) -> ParserFormat:
+        """
+        Returns the ParserFormat based on the file extension
+        """
+        filename = obj["file"].name
+        ext = filename.split('.')[-1]
+        parser_format = getattr(ParserFormat, ext.upper(), None)
+        assert parser_format, f'File type not supported: ${ext.lower()}'
+
+        return parser_format
+
+    def to_internal_value(self, data):
+        """
+        Adds the parser_format to the validated data
+        """
+        validated_data = super().to_internal_value(data)
+        validated_data['parser_format'] = self.get_parser_format(data)
+        return validated_data

openedx_tagging/core/tagging/rest_api/v1/urls.py

@@ -23,4 +23,9 @@
         views_import.TemplateView.as_view(),
         name="taxonomy-import-template",
     ),
+    path(
+        "import/",
+        views_import.ImportView.as_view(),
+        name="taxonomy-import",
+    ),
 ]

openedx_tagging/core/tagging/rest_api/v1/views_import.py

@@ -5,10 +5,14 @@
 
 import os
 
-from django.http import FileResponse, Http404
+from django.http import FileResponse, Http404, HttpResponse, HttpResponseBadRequest
+from rest_framework.exceptions import PermissionDenied
 from rest_framework.request import Request
 from rest_framework.views import APIView
 
+from ...import_export import api
+from .serializers import TaxonomyImportBodySerializer
+
 
 class TemplateView(APIView):
     """
@@ -49,3 +53,51 @@ def get(self, request: Request, file_ext: str, *args, **kwargs) -> FileResponse:
         response = FileResponse(fh, content_type=content_type)
         response['Content-Disposition'] = content_disposition
         return response
+
+
+class ImportView(APIView):
+    """
+    View to import taxonomies
+
+    **Example Requests**
+        POST /tagging/rest_api/v1/import/
+        {
+            "taxonomy_name": "Taxonomy Name",
+            "taxonomy_description": "This is a description",
+            "file": <file>,
+        }
+
+    **Query Returns**
+        * 200 - Success
+        * 400 - Bad request
+        * 405 - Method not allowed
+    """
+    http_method_names = ['post']
+
+    def post(self, request: Request, *args, **kwargs) -> HttpResponse:
+        """
+        Imports the taxonomy from the uploaded file.
+        """
+        perm = "oel_tagging.import_taxonomy"
+        if not request.user.has_perm(perm):
+            raise PermissionDenied("You do not have permission to import taxonomies")
+
+        body = TaxonomyImportBodySerializer(data=request.data)
+        body.is_valid(raise_exception=True)
+
+        taxonomy_name = body.validated_data["taxonomy_name"]
+        taxonomy_description = body.validated_data["taxonomy_description"]
+        file = body.validated_data["file"].file
+        parser_format = body.validated_data["parser_format"]
+
+        import_success = api.create_taxonomy_and_import_tags(
+            taxonomy_name=taxonomy_name,
+            taxonomy_description=taxonomy_description,
+            file=file,
+            parser_format=parser_format,
+        )
+
+        if import_success:
+            return HttpResponse(status=200)
+        else:
+            return HttpResponseBadRequest("Error importing taxonomy")

openedx_tagging/core/tagging/rules.py

@@ -172,6 +172,7 @@ def can_change_object_tag(
 rules.add_perm("oel_tagging.delete_taxonomy", can_change_taxonomy)
 rules.add_perm("oel_tagging.view_taxonomy", can_view_taxonomy)
 rules.add_perm("oel_tagging.export_taxonomy", can_view_taxonomy)
+rules.add_perm("oel_tagging.import_taxonomy", can_change_taxonomy)
 
 # Tag
 rules.add_perm("oel_tagging.add_tag", can_change_tag)

tests/openedx_tagging/core/tagging/test_views_import.py

@@ -3,11 +3,20 @@
 """
 from __future__ import annotations
 
+import json
+
 import ddt  # type: ignore[import]
+from django.contrib.auth import get_user_model
+from django.core.files.uploadedfile import SimpleUploadedFile
 from rest_framework import status
 from rest_framework.test import APITestCase
 
+from openedx_tagging.core.tagging.models import Tag, Taxonomy
+
 TAXONOMY_TEMPLATE_URL = "/tagging/rest_api/v1/import/{filename}"
+TAXONOMY_IMPORT_URL = "/tagging/rest_api/v1/import/"
+
+User = get_user_model()
 
 
 @ddt.ddt
@@ -37,3 +46,181 @@ def test_download_method_not_allowed(self):
         url = TAXONOMY_TEMPLATE_URL.format(filename="template.txt")
         response = self.client.post(url)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
+
+
+@ddt.ddt
+class TestImportView(APITestCase):
+    """
+    Tests the import taxonomy view.
+    """
+    def setUp(self):
+        super().setUp()
+
+        self.user = User.objects.create(
+            username="user",
+            email="user@example.com",
+        )
+
+        self.staff = User.objects.create(
+            username="staff",
+            email="staff@example.com",
+            is_staff=True,
+        )
+
+    def _get_file(self, tags: list, file_format: str) -> SimpleUploadedFile:
+        """
+        Returns a file for the given format.
+        """
+        if file_format == "csv":
+            csv_data = "id,value"
+            for tag in tags:
+                csv_data += f"\n{tag['id']},{tag['value']}"
+            return SimpleUploadedFile("taxonomy.csv", csv_data.encode(), content_type="text/csv")
+        else:  # json
+            json_data = {"tags": tags}
+            return SimpleUploadedFile("taxonomy.json", json.dumps(json_data).encode(), content_type="application/json")
+
+    @ddt.data(
+        "csv",
+        "json",
+    )
+    def test_import(self, file_format: str) -> None:
+        """
+        Tests importing a valid taxonomy file.
+        """
+        url = TAXONOMY_IMPORT_URL
+        new_tags = [
+            {"id": "tag_1", "value": "Tag 1"},
+            {"id": "tag_2", "value": "Tag 2"},
+            {"id": "tag_3", "value": "Tag 3"},
+            {"id": "tag_4", "value": "Tag 4"},
+        ]
+        file = self._get_file(new_tags, file_format)
+
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_name": "Imported Taxonomy name",
+                "taxonomy_description": "Imported Taxonomy description",
+                "file": file,
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        # Check if the taxonomy was created
+        taxonomy = Taxonomy.objects.get(name="Imported Taxonomy name")
+        assert taxonomy.description == "Imported Taxonomy description"
+
+        # Check if the tags were created
+        tags = list(Tag.objects.filter(taxonomy=taxonomy))
+        assert len(tags) == len(new_tags)
+        for i, tag in enumerate(tags):
+            assert tag.value == new_tags[i]["value"]
+
+    def test_import_no_file(self) -> None:
+        """
+        Tests importing a taxonomy without a file.
+        """
+        url = TAXONOMY_IMPORT_URL
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_name": "Imported Taxonomy name",
+                "taxonomy_description": "Imported Taxonomy description",
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["file"][0] == "No file was submitted."
+
+    @ddt.data(
+        "csv",
+        "json",
+    )
+    def test_import_no_name(self, file_format) -> None:
+        """
+        Tests importing a taxonomy without specifing a name.
+        """
+        url = TAXONOMY_IMPORT_URL
+        file = SimpleUploadedFile(f"taxonomy.{file_format}", b"invalid file content")
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_description": "Imported Taxonomy description",
+                "file": file,
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["taxonomy_name"][0] == "This field is required."
+
+    def test_import_invalid_format(self) -> None:
+        """
+        Tests importing a taxonomy with an invalid file format.
+        """
+        url = TAXONOMY_IMPORT_URL
+        file = SimpleUploadedFile("taxonomy.invalid", b"invalid file content")
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_name": "Imported Taxonomy name",
+                "taxonomy_description": "Imported Taxonomy description",
+                "file": file,
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["file"][0] == "File type not supported: invalid"
+
+    @ddt.data(
+        "csv",
+        "json",
+    )
+    def test_import_invalid_content(self, file_format) -> None:
+        """
+        Tests importing a taxonomy with an invalid file content.
+        """
+        url = TAXONOMY_IMPORT_URL
+        file = SimpleUploadedFile(f"taxonomy.{file_format}", b"invalid file content")
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_name": "Imported Taxonomy name",
+                "taxonomy_description": "Imported Taxonomy description",
+                "file": file,
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.content == b"Error importing taxonomy"
+
+    def test_import_no_perm(self) -> None:
+        """
+        Tests importing a taxonomy using a user without permission.
+        """
+        url = TAXONOMY_IMPORT_URL
+        new_tags = [
+            {"id": "tag_1", "value": "Tag 1"},
+            {"id": "tag_2", "value": "Tag 2"},
+            {"id": "tag_3", "value": "Tag 3"},
+            {"id": "tag_4", "value": "Tag 4"},
+        ]
+        file = self._get_file(new_tags, "json")
+
+        self.client.force_authenticate(user=self.user)
+        response = self.client.post(
+            url,
+            {
+                "taxonomy_name": "Imported Taxonomy name",
+                "taxonomy_description": "Imported Taxonomy description",
+                "file": file,
+            },
+            format="multipart"
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN