Simplify Tag Models [FC-0030]

openedx_tagging/core/tagging/api.py

@@ -14,11 +14,15 @@
 
 from typing import Iterator
 
-from django.db.models import QuerySet
+from django.db import transaction
+from django.db.models import F, QuerySet
 from django.utils.translation import gettext_lazy as _
 
 from .models import ObjectTag, Tag, Taxonomy
 
+# Export this as part of the API
+TagDoesNotExist = Tag.DoesNotExist
+
 
 def create_taxonomy(
     name: str,
@@ -139,21 +143,18 @@ def resync_object_tags(object_tags: QuerySet | None = None) -> int:
 
 
 def get_object_tags(
-    object_id: str, taxonomy_id: str | None = None
+    object_id: str,
+    taxonomy_id: int | None = None,
+    object_tag_class: type[ObjectTag] = ObjectTag
 ) -> QuerySet[ObjectTag]:
     """
     Returns a Queryset of object tags for a given object.
 
-    Pass taxonomy to limit the returned object_tags to a specific taxonomy.
+    Pass taxonomy_id to limit the returned object_tags to a specific taxonomy.
     """
-    ObjectTagClass = ObjectTag
-    extra_filters = {}
-    if taxonomy_id is not None:
-        taxonomy = Taxonomy.objects.get(pk=taxonomy_id)
-        ObjectTagClass = taxonomy.object_tag_class
-        extra_filters["taxonomy_id"] = taxonomy_id
+    filters = {"taxonomy_id": taxonomy_id} if taxonomy_id else {}
     tags = (
-        ObjectTagClass.objects.filter(object_id=object_id, **extra_filters)
+        object_tag_class.objects.filter(object_id=object_id, **filters)
         .select_related("tag", "taxonomy")
         .order_by("id")
     )
@@ -164,30 +165,103 @@ def delete_object_tags(object_id: str):
     """
     Delete all ObjectTag entries for a given object.
     """
-    tags = ObjectTag.objects.filter(
-        object_id=object_id,
-    )
+    tags = ObjectTag.objects.filter(object_id=object_id)
 
     tags.delete()
 
 
+# TODO: a function called "tag_object" should take "object_id" as its first parameter, not taxonomy
 def tag_object(
     taxonomy: Taxonomy,
     tags: list[str],
     object_id: str,
-) -> list[ObjectTag]:
+    object_tag_class: type[ObjectTag] = ObjectTag,
+) -> None:
     """
-    Replaces the existing ObjectTag entries for the given taxonomy + object_id with the given list of tags.
+    Replaces the existing ObjectTag entries for the given taxonomy + object_id
+    with the given list of tags.
 
-    If taxonomy.allows_free_text, then the list should be a list of tag values.
-    Otherwise, it should be a list of existing Tag IDs.
+    tags: A list of the values of the tags from this taxonomy to apply.
 
-    Raised ValueError if the proposed tags are invalid for this taxonomy.
-    Preserves existing (valid) tags, adds new (valid) tags, and removes omitted (or invalid) tags.
+    object_tag_class: Optional. Use a proxy subclass of ObjectTag for additional
+        validation. (e.g. only allow tagging certain types of objects.)
+
+    Raised Tag.DoesNotExist if the proposed tags are invalid for this taxonomy.
+    Preserves existing (valid) tags, adds new (valid) tags, and removes omitted
+    (or invalid) tags.
     """
-    return taxonomy.cast().tag_object(tags, object_id)
 
+    def _check_new_tag_count(new_tag_count: int) -> None:
+        """
+        Checks if the new count of tags for the object is equal or less than 100
+        """
+        # Exclude self.id to avoid counting the tags that are going to be updated
+        current_count = ObjectTag.objects.filter(object_id=object_id).exclude(taxonomy_id=taxonomy.id).count()
+
+        if current_count + new_tag_count > 100:
+            raise ValueError(
+                _(f"Cannot add more than 100 tags to ({object_id}).")
+            )
+
+    if not isinstance(tags, list):
+        raise ValueError(_(f"Tags must be a list, not {type(tags).__name__}."))
+
+    ObjectTagClass = object_tag_class
+    taxonomy = taxonomy.cast()  # Make sure we're using the right subclass. This is a no-op if we are already.
+    tags = list(dict.fromkeys(tags))  # Remove duplicates preserving order
+
+    _check_new_tag_count(len(tags))
+
+    if not taxonomy.allow_multiple and len(tags) > 1:
+        raise ValueError(_(f"Taxonomy ({taxonomy.name}) 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 = list(
+        ObjectTagClass.objects.filter(taxonomy=taxonomy, object_id=object_id)
+    )
+    updated_tags = []
+    if taxonomy.allow_free_text:
+        for tag_value in tags:
+            object_tag_index = next((i for (i, t) in enumerate(current_tags) if t._value == tag_value), -1)
+            if object_tag_index >= 0:
+                # This tag is already applied.
+                object_tag = current_tags.pop(object_tag_index)
+            else:
+                object_tag = ObjectTagClass(taxonomy=taxonomy, object_id=object_id, _value=tag_value)
+                updated_tags.append(object_tag)
+    else:
+        # Handle closed taxonomies:
+        for tag_value in tags:
+            tag = taxonomy.tag_for_value(tag_value)  # Will raise Tag.DoesNotExist if the value is invalid.
+            object_tag_index = next((i for (i, t) in enumerate(current_tags) if t.tag_id == tag.id), -1)
+            if object_tag_index >= 0:
+                # This tag is already applied.
+                object_tag = current_tags.pop(object_tag_index)
+                if object_tag._value != tag.value:
+                    # The ObjectTag's cached '_value' is out of sync with the Tag, so update it:
+                    object_tag._value = tag.value
+                    updated_tags.append(object_tag)
+            else:
+                # We are newly applying this tag:
+                object_tag = ObjectTagClass(taxonomy=taxonomy, object_id=object_id, tag=tag)
+                updated_tags.append(object_tag)
+
+    # Save all updated tags at once to avoid partial updates
+    with transaction.atomic():
+        # delete any omitted existing tags. We do this first to reduce chances of UNIQUE constraint edge cases
+        for old_tag in current_tags:
+            old_tag.delete()
+        # add the new tags:
+        for object_tag in updated_tags:
+            object_tag.full_clean()  # Run validation
+            object_tag.save()
+
+
+# TODO: return tags from closed taxonomies as well as the count of how many times each is used.
 def autocomplete_tags(
     taxonomy: Taxonomy,
     search: str,
@@ -228,4 +302,25 @@ def autocomplete_tags(
                 "using get_tags() and filtering them on the frontend."
             )
         )
-    return taxonomy.cast().autocomplete_tags(search, object_id)
+    # Fetch tags that the object already has to exclude them from the result
+    excluded_tags: list[str] = []
+    if object_id:
+        excluded_tags = list(
+            taxonomy.objecttag_set.filter(object_id=object_id).values_list(
+                "_value", flat=True
+            )
+        )
+    return (
+        # Fetch object tags from this taxonomy whose value contains the search
+        taxonomy.objecttag_set.filter(_value__icontains=search)
+        # omit any tags whose values match the tags on the given object
+        .exclude(_value__in=excluded_tags)
+        # alphabetical ordering
+        .order_by("_value")
+        # Alias the `_value` field to `value` to make it nicer for users
+        .annotate(value=F("_value"))
+        # obtain tag values
+        .values("value", "tag_id")
+        # remove repeats
+        .distinct()
+    )

openedx_tagging/core/tagging/fixtures/language_taxonomy.yaml

@@ -1296,3 +1296,4 @@
     allow_multiple: false
     allow_free_text: false
     visible_to_authors: true
+    _taxonomy_class: openedx_tagging.core.tagging.models.system_defined.LanguageTaxonomy

openedx_tagging/core/tagging/migrations/0010_cleanups.py

@@ -0,0 +1,32 @@
+# Generated by Django 3.2.19 on 2023-09-29 16:59
+
+import django.db.models.expressions
+from django.db import migrations, models
+
+import openedx_learning.lib.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('oel_tagging', '0009_alter_objecttag_object_id'),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name='ModelObjectTag',
+        ),
+        migrations.DeleteModel(
+            name='UserModelObjectTag',
+        ),
+        migrations.AlterUniqueTogether(
+            name='objecttag',
+            unique_together={('object_id', 'taxonomy', 'tag_id'), ('object_id', 'taxonomy', '_value')},
+        ),
+        # ObjectTag.Tag can be blank
+        migrations.AlterField(
+            model_name='objecttag',
+            name='tag',
+            field=models.ForeignKey(blank=True, default=None, help_text="Tag associated with this object tag. Provides the tag's 'value' if set.", null=True, on_delete=django.db.models.deletion.SET_NULL, to='oel_tagging.tag'),
+        ),
+    ]

openedx_tagging/core/tagging/models/__init__.py

@@ -1,3 +1,3 @@
 from .base import ObjectTag, Tag, Taxonomy
 from .import_export import TagImportTask, TagImportTaskState
-from .system_defined import LanguageTaxonomy, ModelObjectTag, ModelSystemDefinedTaxonomy, UserSystemDefinedTaxonomy
+from .system_defined import LanguageTaxonomy, ModelSystemDefinedTaxonomy, UserSystemDefinedTaxonomy