Allow custom Taxonomy, ObjectTag subclasses to customize tagging behavior

openedx_tagging/core/tagging/api.py

@@ -10,33 +10,48 @@
 Please look at the models.py file for more information about the kinds of data
 are stored in this app.
 """
-from typing import List, Type
+from typing import Generator, List, Type, Union
 
 from django.db.models import QuerySet
 from django.utils.translation import gettext_lazy as _
 
-from .models import ObjectTag, Tag, Taxonomy
+from .models import ClosedObjectTag, ObjectTag, OpenObjectTag, Tag, Taxonomy
+from .registry import get_object_tag_class
 
 
 def create_taxonomy(
-    name,
-    description=None,
+    name: str,
+    description: str = None,
     enabled=True,
     required=False,
     allow_multiple=False,
     allow_free_text=False,
+    system_defined=False,
+    object_tag_class: Type = None,
 ) -> Taxonomy:
     """
     Creates, saves, and returns a new Taxonomy with the given attributes.
     """
-    return Taxonomy.objects.create(
+    taxonomy = Taxonomy(
         name=name,
         description=description,
         enabled=enabled,
         required=required,
         allow_multiple=allow_multiple,
         allow_free_text=allow_free_text,
+        system_defined=system_defined,
     )
+    if object_tag_class:
+        taxonomy.object_tag_class = object_tag_class
+    taxonomy.save()
+    return taxonomy
+
+
+def get_taxonomy(id: int) -> Union[Taxonomy, None]:
+    """
+    Returns a Taxonomy of the appropriate subclass which has the given ID.
+    """
+    return Taxonomy.objects.filter(id=id).first()
 
 
 def get_taxonomies(enabled=True) -> QuerySet:
@@ -60,17 +75,37 @@ def get_tags(taxonomy: Taxonomy) -> List[Tag]:
     return taxonomy.get_tags()
 
 
+def cast_object_tag(object_tag: ObjectTag) -> ObjectTag:
+    """
+    Casts/copies the given object tag data into the ObjectTag subclass most appropriate for this tag.
+
+    E.g. if the tag's taxonomy has a custom ObjectTag class it prefers, use that.
+    Recommends using OpenObjectTag by default, because that is the least restrictive type.
+    """
+    ObjectTagClass = get_object_tag_class(
+        taxonomy=object_tag.taxonomy,
+        object_type=object_tag.object_type,
+        object_id=object_tag.object_id,
+        tag=object_tag.tag,
+        value=object_tag.value,
+        name=object_tag.name,
+    )
+    new_object_tag = ObjectTagClass().copy(object_tag)
+    return new_object_tag
+
+
 def resync_object_tags(object_tags: QuerySet = None) -> int:
     """
     Reconciles ObjectTag entries with any changes made to their associated taxonomies and tags.
 
     By default, we iterate over all ObjectTags. Pass a filtered ObjectTags queryset to limit which tags are resynced.
     """
     if not object_tags:
-        object_tags = ObjectTag.objects.all()
+        object_tags = ObjectTag.objects.select_related("tag", "taxonomy")
 
     num_changed = 0
-    for object_tag in object_tags:
+    for tag in object_tags:
+        object_tag = cast_object_tag(tag)
         changed = object_tag.resync()
         if changed:
             object_tag.save()
@@ -79,18 +114,33 @@ def resync_object_tags(object_tags: QuerySet = None) -> int:
 
 
 def get_object_tags(
-    taxonomy: Taxonomy, object_id: str, object_type: str, valid_only=True
-) -> List[ObjectTag]:
+    object_id: str, object_type: str = None, taxonomy: Taxonomy = None, valid_only=True
+) -> Generator[ObjectTag, None, None]:
     """
-    Returns a list of tags for a given taxonomy + content.
+    Generates a list of object tags for a given object.
+
+    Pass taxonomy to limit the returned object_tags to a specific taxonomy.
 
     Pass valid_only=False when displaying tags to content authors, so they can see invalid tags too.
-    Invalid tags will likely be hidden from learners.
+    Invalid tags will (probably) be hidden from learners.
     """
-    tags = ObjectTag.objects.filter(
-        taxonomy=taxonomy, object_id=object_id, object_type=object_type
-    ).order_by("id")
-    return [tag for tag in tags if not valid_only or taxonomy.validate_object_tag(tag)]
+    tags = (
+        ObjectTag.objects.filter(
+            object_id=object_id,
+        )
+        .select_related("tag", "taxonomy")
+        .order_by("id")
+    )
+    if object_type:
+        tags = tags.filter(object_type=object_type)
+    if taxonomy:
+        tags = tags.filter(taxonomy=taxonomy)
+
+    for tag in tags:
+        # We can only validate tags with taxonomies, because we need the object_tag_class
+        object_tag = cast_object_tag(tag)
+        if not valid_only or object_tag.is_valid():
+            yield object_tag
 
 
 def tag_object(
@@ -106,4 +156,64 @@ def tag_object(
     Preserves existing (valid) tags, adds new (valid) tags, and removes omitted (or invalid) tags.
     """
 
-    return taxonomy.tag_object(tags, object_id, object_type)
+    if not taxonomy.allow_multiple and len(tags) > 1:
+        raise ValueError(_(f"Taxonomy ({taxonomy.id}) only allows one tag per object."))
+
+    if taxonomy.required and len(tags) == 0:
+        raise ValueError(
+            _(f"Taxonomy ({taxonomy.id}) requires at least one tag per object.")
+        )
+
+    current_tags = {
+        tag.tag_ref: tag
+        for tag in ObjectTag.objects.filter(
+            taxonomy=taxonomy, object_id=object_id, object_type=object_type
+        )
+    }
+    updated_tags = []
+    for tag_ref in tags:
+        if tag_ref in current_tags:
+            object_tag = cast_object_tag(current_tags.pop(tag_ref))
+        else:
+            try:
+                tag = taxonomy.tag_set.get(
+                    id=tag_ref,
+                )
+                value = tag.value
+            except (ValueError, Tag.DoesNotExist):
+                # This might be ok, e.g. if taxonomy.allow_free_text.
+                # We'll validate below before saving.
+                tag = None
+                value = tag_ref
+
+            ObjectTagClass = get_object_tag_class(
+                taxonomy=taxonomy,
+                object_id=object_id,
+                object_type=object_type,
+                tag=tag,
+                value=value,
+                name=taxonomy.name,
+            )
+            object_tag = ObjectTagClass()
+            object_tag.taxonomy = taxonomy
+            object_tag.tag = tag
+            object_tag.object_id = object_id
+            object_tag.object_type = object_type
+            object_tag.value = value
+
+        object_tag.resync()
+        if not object_tag.is_valid():
+            raise ValueError(
+                _(f"Invalid object tag for taxonomy ({taxonomy.id}): {object_tag}")
+            )
+        updated_tags.append(object_tag)
+
+    # Save all updated tags at once to avoid partial updates
+    for object_tag in updated_tags:
+        object_tag.save()
+
+    # ...and delete any omitted existing tags
+    for old_tag in current_tags.values():
+        old_tag.delete()
+
+    return updated_tags

openedx_tagging/core/tagging/migrations/0002_taxonomy_system_defined.py

@@ -0,0 +1,20 @@
+# Generated by Django 3.2.19 on 2023-07-05 04:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("oel_tagging", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="taxonomy",
+            name="system_defined",
+            field=models.BooleanField(
+                default=False,
+                help_text="Indicates that tags and metadata for this taxonomy are maintained by the system; taxonomy admins will not be permitted to modify them.",
+            ),
+        ),
+    ]

openedx_tagging/core/tagging/migrations/0003_objecttag_proxies.py

@@ -0,0 +1,41 @@
+# Generated by Django 3.2.19 on 2023-07-05 05:07
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("oel_tagging", "0002_taxonomy_system_defined"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="OpenObjectTag",
+            fields=[],
+            options={
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("oel_tagging.objecttag",),
+        ),
+        migrations.AddField(
+            model_name="taxonomy",
+            name="_object_tag_class",
+            field=models.CharField(
+                help_text="Overrides the default ObjectTag subclass associated with this taxonomy.Must be a fully-qualified module and class name.",
+                max_length=255,
+                null=True,
+            ),
+        ),
+        migrations.CreateModel(
+            name="ClosedObjectTag",
+            fields=[],
+            options={
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("oel_tagging.openobjecttag",),
+        ),
+    ]

openedx_tagging/core/tagging/migrations/0004_tag_cascade_delete.py

@@ -0,0 +1,36 @@
+# Generated by Django 3.2.19 on 2023-07-06 03:56
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("oel_tagging", "0003_objecttag_proxies"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="tag",
+            name="parent",
+            field=models.ForeignKey(
+                default=None,
+                help_text="Tag that lives one level up from the current tag, forming a hierarchy.",
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="children",
+                to="oel_tagging.tag",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="tag",
+            name="taxonomy",
+            field=models.ForeignKey(
+                default=None,
+                help_text="Namespace and rules for using a given set of tags.",
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                to="oel_tagging.taxonomy",
+            ),
+        ),
+    ]