feat: export functions

openedx_tagging/core/tagging/import_export/api.py

@@ -1,3 +1,5 @@
+"""
+"""
 from io import BytesIO
 
 from django.utils.translation import gettext_lazy as _
@@ -13,6 +15,8 @@ def import_tags(
     parser_format: ParserFormat,
     replace=False,
 ) -> bool:
+    _import_export_validations(taxonomy)
+
     # Checks that exists only one tag
     if not _check_unique_import_task(taxonomy):
         raise ValueError(
@@ -68,6 +72,12 @@ def get_last_import_log(taxonomy: Taxonomy) -> str:
     return task.log
 
 
+def export_tags(taxonomy: Taxonomy, output_format: ParserFormat) -> str:
+    _import_export_validations(taxonomy)
+    parser = get_parser(output_format)
+    return parser.export(taxonomy)
+
+
 def _check_unique_import_task(taxonomy: Taxonomy) -> bool:
     last_task = _get_last_tags(taxonomy)
     if not last_task:
@@ -84,3 +94,19 @@ def _get_last_tags(taxonomy: Taxonomy) -> TagImportTask:
         .order_by("-creation_date")
         .first()
     )
+
+
+def _import_export_validations(taxonomy: Taxonomy):
+    taxonomy = taxonomy.cast()
+    if taxonomy.allow_free_text:
+        raise ValueError(
+            _(
+                f"Invalid taxonomy ({taxonomy.id}): You cannot import/export a free-form taxonomy."
+            )
+        )
+    if taxonomy.system_defined:
+        raise ValueError(
+            _(
+                f"Invalid taxonomy ({taxonomy.id}): You cannot import/export a system-defined taxonomy."
+            )
+        )

openedx_tagging/core/tagging/import_export/parsers.py

@@ -4,7 +4,7 @@
 import csv
 import json
 from enum import Enum
-from io import BytesIO, TextIOWrapper
+from io import BytesIO, TextIOWrapper, StringIO
 from typing import List, Tuple
 
 from django.utils.translation import gettext_lazy as _
@@ -17,6 +17,8 @@
     EmptyJSONField,
     EmptyCSVField,
 )
+from ..models import Taxonomy
+from ..api import get_tags
 
 
 class ParserFormat(Enum):
@@ -51,8 +53,10 @@ class Parser:
     @classmethod
     def parse_import(cls, file: BytesIO) -> Tuple[List[TagDSL], List[TagParserError]]:
         """
+        Parse tags in file an returns tags ready for use in TagImportPlan
+
         Top function that calls `_load_data` and `_parse_tags`.
-        Handle the errors returned both functions
+        Handle errors returned by both functions.
         """
         try:
             tags_data, load_errors = cls._load_data(file)
@@ -65,6 +69,15 @@ def parse_import(cls, file: BytesIO) -> Tuple[List[TagDSL], List[TagParserError]
 
         return cls._parse_tags(tags_data)
 
+    @classmethod
+    def export(cls, taxonomy: Taxonomy) -> str:
+        """
+        Returns all tags in taxonomy.
+        The output file can be used to recreate the taxonomy with `parse_import`
+        """
+        tags = cls._load_tags_for_export(taxonomy)
+        return cls._export_data(tags, taxonomy)
+
     @classmethod
     def _load_data(cls, file: BytesIO) -> Tuple[List[dict], List[TagParserError]]:
         """
@@ -76,6 +89,18 @@ def _load_data(cls, file: BytesIO) -> Tuple[List[dict], List[TagParserError]]:
         """
         raise NotImplementedError
 
+    @classmethod
+    def _export_data(cls, tags: List[dict], taxonomy: Taxonomy) -> str:
+        """
+        Each parser implements this function according to its format.
+        Returns a string with tags data in the parser format.
+        Can use `taxonomy` to export taxonomy metadata.
+
+        It must be implemented in such a way that the output of
+        this function works with _load_data
+        """
+        raise NotImplementedError
+
     @classmethod
     def _parse_tags(cls, tags_data: dict) -> Tuple[List[TagDSL], List[TagParserError]]:
         """
@@ -129,6 +154,30 @@ def _parse_tags(cls, tags_data: dict) -> Tuple[List[TagDSL], List[TagParserError
 
         return tags, errors
 
+    @classmethod
+    def _load_tags_for_export(cls, taxonomy: Taxonomy) -> List[dict]:
+        """
+        Returns a list of taxonomy's tags in the form of a dictionary
+        with required and optional fields
+
+        The 'action' field is not added to the result because the output
+        is seen as a creation of all the tags.
+
+        The tags are ordered by hierarchy, first, parents and then children.
+        `get_tags` is in charge of returning this in a hierarchical way.
+        """
+        tags = get_tags(taxonomy)
+        result = []
+        for tag in tags:
+            result_tag = {
+                "id": tag.external_id or tag.id,
+                "value": tag.value,
+            }
+            if tag.parent:
+                result_tag["parent_id"] = tag.parent.external_id or tag.parent.id
+            result.append(result_tag)
+        return result
+
 
 class JSONParser(Parser):
     """
@@ -172,6 +221,18 @@ def _load_data(cls, file: BytesIO) -> Tuple[List[dict], List[TagParserError]]:
         tags_data = tags_data.get("tags")
         return tags_data, []
 
+    @classmethod
+    def _export_data(cls, tags: List[dict], taxonomy: Taxonomy) -> str:
+        """
+        Export tags and taxonomy metadata in JSON format
+        """
+        json_result = {
+            "name": taxonomy.name,
+            "description": taxonomy.description,
+            "tags": tags,
+        }
+        return json.dumps(json_result)
+
 
 class CSVParser(Parser):
     """
@@ -203,6 +264,28 @@ def _load_data(cls, file: BytesIO) -> Tuple[List[dict], List[TagParserError]]:
             return None, errors
         return list(csv_reader), []
 
+    @classmethod
+    def _export_data(cls, tags: List[dict], taxonomy: Taxonomy) -> str:
+        """
+        Export tags in CSV format
+
+        The 'action' field is not added to the result because the output
+        is seen as a creation of all the tags.
+        """
+        fields = cls.required_fields + cls.optional_fields
+        if "action" in fields:  # pragma: no cover
+            fields.remove("action")
+
+        with StringIO() as csv_buffer:
+            csv_writer = csv.DictWriter(csv_buffer, fieldnames=fields)
+            csv_writer.writeheader()
+
+            for tag in tags:
+                csv_writer.writerow(tag)
+
+            csv_string = csv_buffer.getvalue()
+            return csv_string
+
     @classmethod
     def _veify_header(cls, header_fields: List[str]) -> List[TagParserError]:
         """

openedx_tagging/core/tagging/migrations/0005_language_taxonomy.py

@@ -11,7 +11,7 @@ def load_language_taxonomy(apps, schema_editor):
     call_command("loaddata", "--app=oel_tagging", "language_taxonomy.yaml")
 
 
-def revert(apps, schema_editor):
+def revert(apps, schema_editor):  # pragma: no cover
     """
     Deletes language taxonomy an tags
     """

tests/openedx_tagging/core/tagging/import_export/mixins.py

@@ -0,0 +1,16 @@
+"""
+Mixins for ImportExport tests
+"""
+from openedx_tagging.core.tagging.models import Taxonomy
+
+
+class TestImportExportMixin:
+    """
+    Mixin that loads the base data for import/export tests
+    """
+
+    fixtures = ["tests/openedx_tagging/core/fixtures/tagging.yaml"]
+
+    def setUp(self):
+        self.taxonomy = Taxonomy.objects.get(name="Import Taxonomy Test")
+        return super().setUp()

tests/openedx_tagging/core/tagging/import_export/test_actions.py

@@ -5,7 +5,7 @@
 
 from django.test.testcases import TestCase
 
-from openedx_tagging.core.tagging.models import Taxonomy, Tag
+from openedx_tagging.core.tagging.models import Tag
 from openedx_tagging.core.tagging.import_export.import_plan import TagDSL
 from openedx_tagging.core.tagging.import_export.actions import (
     ImportAction,
@@ -15,17 +15,15 @@
     DeleteTag,
     WithoutChanges,
 )
+from .mixins import TestImportExportMixin
 
 
-class TestImportActionMixin:
+class TestImportActionMixin(TestImportExportMixin):
     """
     Mixin for import action tests
     """
-
-    fixtures = ["tests/openedx_tagging/core/fixtures/tagging.yaml"]
-
     def setUp(self):
-        self.taxonomy = Taxonomy.objects.get(name="Import Taxonomy Test")
+        super().setUp()
         self.indexed_actions = {
             'create': [
                 CreateTag(

tests/openedx_tagging/core/tagging/import_export/test_api.py

@@ -6,16 +6,21 @@
 
 from django.test.testcases import TestCase
 
-from openedx_tagging.core.tagging.models import TagImportTask, TagImportTaskState
+from openedx_tagging.core.tagging.models import (
+    TagImportTask,
+    TagImportTaskState,
+    Taxonomy,
+    LanguageTaxonomy,
+)
 from openedx_tagging.core.tagging.import_export import ParserFormat
 import openedx_tagging.core.tagging.import_export.api as import_export_api
 
-from .test_actions import TestImportActionMixin
+from .mixins import TestImportExportMixin
 
 
-class TestImportApi(TestImportActionMixin, TestCase):
+class TestImportExportApi(TestImportExportMixin, TestCase):
     """
-    Test import API functions
+    Test import/export API functions
     """
 
     def setUp(self):
@@ -40,6 +45,16 @@ def setUp(self):
         self.invalid_plan_file = BytesIO(json.dumps(json_data).encode())
 
         self.parser_format = ParserFormat.JSON
+
+        self.open_taxonomy = Taxonomy(
+            name="Open taxonomy",
+            allow_free_text=True
+        )
+        self.system_taxonomy = Taxonomy(
+            name="System taxonomy",
+        )
+        self.system_taxonomy.taxonomy_class = LanguageTaxonomy
+        self.system_taxonomy = self.system_taxonomy.cast()
         return super().setUp()
 
     def test_check_status(self):
@@ -62,6 +77,23 @@ def test_invalid_import_tags(self):
                 self.parser_format,
             )
 
+    def test_import_export_validations(self):
+        # Check that import is invalid with open taxonomy
+        with self.assertRaises(ValueError):
+            import_export_api.import_tags(
+                self.open_taxonomy,
+                self.file,
+                self.parser_format,
+            )
+
+        # Check that import is invalid with system taxonomy
+        with self.assertRaises(ValueError):
+            import_export_api.import_tags(
+                self.system_taxonomy,
+                self.file,
+                self.parser_format,
+            )
+
     def test_with_python_error(self):
         self.file.close()
         assert not import_export_api.import_tags(
@@ -146,3 +178,42 @@ def test_start_task_after_success(self):
             self.file,
             self.parser_format,
         )
+
+    def test_export_validations(self):
+        # Check that import is invalid with open taxonomy
+        with self.assertRaises(ValueError):
+            import_export_api.export_tags(
+                self.open_taxonomy,
+                self.parser_format,
+            )
+
+        # Check that import is invalid with system taxonomy
+        with self.assertRaises(ValueError):
+            import_export_api.export_tags(
+                self.system_taxonomy,
+                self.parser_format,
+            )
+
+    def test_import_with_export_output(self):
+        for parser_format in ParserFormat:
+            output = import_export_api.export_tags(
+                self.taxonomy,
+                parser_format,
+            )
+            file = BytesIO(output.encode())
+            new_taxonomy = Taxonomy(name="New taxonomy")
+            new_taxonomy.save()
+            assert import_export_api.import_tags(
+                new_taxonomy,
+                file,
+                parser_format,
+            )
+            old_tags = self.taxonomy.tag_set.all()
+            assert len(old_tags) == new_taxonomy.tag_set.count()
+
+            for tag in old_tags:
+                new_tag = new_taxonomy.tag_set.get(external_id=tag.external_id)
+                assert new_tag.value == tag.value
+                if tag.parent:
+                    assert tag.parent.external_id == new_tag.parent.external_id
+