apps/newtagging/models.py

   1 # -*- coding: utf-8 -*-
   2 """
   3 Models and managers for generic tagging.
   4 """
   5
   6 from django.contrib.contenttypes import generic
   7 from django.contrib.contenttypes.models import ContentType
   8 from django.db import connection, models
   9 from django.utils.translation import ugettext_lazy as _
  10 from django.db.models.base import ModelBase
  11 from django.db.models.loading import get_model # D1.7: apps?
  12 from django.core.exceptions import ObjectDoesNotExist
  13 from django.dispatch import Signal
  14
  15 qn = connection.ops.quote_name
  16
  17
  18 tags_updated = Signal(providing_args=["affected_tags"])
  19
  20 def get_queryset_and_model(queryset_or_model):
  21     """
  22     Given a ``QuerySet`` or a ``Model``, returns a two-tuple of
  23     (queryset, model).
  24
  25     If a ``Model`` is given, the ``QuerySet`` returned will be created
  26     using its default manager.
  27     """
  28     try:
  29         return queryset_or_model, queryset_or_model.model
  30     except AttributeError:
  31         return queryset_or_model._default_manager.all(), queryset_or_model
  32
  33
  34 ############
  35 # Managers #
  36 ############
  37 class TagManager(models.Manager):
  38     def __init__(self, intermediary_table_model):
  39         super(TagManager, self).__init__()
  40         self.intermediary_table_model = intermediary_table_model
  41         models.signals.pre_delete.connect(self.target_deleted)
  42
  43     def target_deleted(self, instance, **kwargs):
  44         """ clear tag relations before deleting an object """
  45         try:
  46             int(instance.pk)
  47         except ValueError:
  48             return
  49
  50         self.update_tags(instance, [])
  51
  52     def update_tags(self, obj, tags):
  53         """
  54         Update tags associated with an object.
  55         """
  56         content_type = ContentType.objects.get_for_model(obj)
  57         current_tags = list(self.filter(items__content_type__pk=content_type.pk,
  58                                         items__object_id=obj.pk))
  59         updated_tags = self.model.get_tag_list(tags)
  60
  61         # Remove tags which no longer apply
  62         tags_for_removal = [tag for tag in current_tags \
  63                             if tag not in updated_tags]
  64         if len(tags_for_removal):
  65             self.intermediary_table_model._default_manager.filter(content_type__pk=content_type.pk,
  66                                                object_id=obj.pk,
  67                                                tag__in=tags_for_removal).delete()
  68         # Add new tags
  69         tags_to_add = [tag for tag in updated_tags
  70                        if tag not in current_tags]
  71         for tag in tags_to_add:
  72             if tag not in current_tags:
  73                 self.intermediary_table_model._default_manager.create(tag=tag, content_object=obj)
  74
  75         tags_updated.send(sender=obj, affected_tags=tags_to_add + tags_for_removal)
  76
  77     def remove_tag(self, obj, tag):
  78         """
  79         Remove tag from an object.
  80         """
  81         content_type = ContentType.objects.get_for_model(obj)
  82         self.intermediary_table_model._default_manager.filter(content_type__pk=content_type.pk,
  83             object_id=obj.pk, tag=tag).delete()
  84
  85     def get_for_object(self, obj):
  86         """
  87         Create a queryset matching all tags associated with the given
  88         object.
  89         """
  90         ctype = ContentType.objects.get_for_model(obj)
  91         return self.filter(items__content_type__pk=ctype.pk,
  92                            items__object_id=obj.pk)
  93
  94     def usage_for_model(self, model, counts=False, filters=None):
  95         """
  96         Obtain a list of tags associated with instances of the given
  97         Model class.
  98
  99         If ``counts`` is True, a ``count`` attribute will be added to
 100         each tag, indicating how many times it has been used against
 101         the Model class in question.
 102
 103         To limit the tags (and counts, if specified) returned to those
 104         used by a subset of the Model's instances, pass a dictionary
 105         of field lookups to be applied to the given Model as the
 106         ``filters`` argument.
 107         """
 108         if filters is None: filters = {}
 109
 110         queryset = model._default_manager.filter()
 111         for f in filters.items():
 112             queryset.query.add_filter(f)
 113         usage = self.usage_for_queryset(queryset, counts)
 114         return usage
 115
 116     def usage_for_queryset(self, queryset, counts=False):
 117         """
 118         Obtain a list of tags associated with instances of a model
 119         contained in the given queryset.
 120
 121         If ``counts`` is True, a ``count`` attribute will be added to
 122         each tag, indicating how many times it has been used against
 123         the Model class in question.
 124         """
 125         usage = self.model._default_manager.filter(
 126             items__content_type=ContentType.objects.get_for_model(queryset.model),
 127             items__object_id__in=queryset
 128             )
 129         if counts:
 130             usage = usage.annotate(count=models.Count('id'))
 131         else:
 132             usage = usage.distinct()
 133         return usage
 134
 135     def related_for_model(self, tags, model, counts=False):
 136         """
 137         Obtain a list of tags related to a given list of tags - that
 138         is, other tags used by items which have all the given tags.
 139
 140         If ``counts`` is True, a ``count`` attribute will be added to
 141         each tag, indicating the number of items which have it in
 142         addition to the given list of tags.
 143         """
 144         objs = self.model.intermediary_table_model.objects.get_by_model(model, tags)
 145         qs = self.usage_for_queryset(objs, counts)
 146         qs = qs.exclude(pk__in=[tag.pk for tag in tags])
 147         return qs
 148
 149
 150 class TaggedItemManager(models.Manager):
 151     def __init__(self, tag_model):
 152         super(TaggedItemManager, self).__init__()
 153         self.tag_model = tag_model
 154
 155     def get_by_model(self, queryset_or_model, tags):
 156         """
 157         Create a ``QuerySet`` containing instances of the specified
 158         model associated with a given tag or list of tags.
 159         """
 160         queryset, model = get_queryset_and_model(queryset_or_model)
 161         tags = self.tag_model.get_tag_list(tags)
 162         tag_count = len(tags)
 163         if not tag_count:
 164             # No existing tags were given
 165             return queryset
 166         elif tag_count == 1:
 167             # Optimisation for single tag - fall through to the simpler
 168             # query below.
 169             return queryset.filter(tag_relations__tag=tags[0])
 170
 171         # TODO: presumes reverse generic relation
 172         return queryset.filter(tag_relations__tag__in=tags
 173             ).annotate(count=models.Count('pk')).filter(count=len(tags))
 174
 175     def get_union_by_model(self, queryset_or_model, tags):
 176         """
 177         Create a ``QuerySet`` containing instances of the specified
 178         model associated with *any* of the given list of tags.
 179         """
 180         queryset, model = get_queryset_and_model(queryset_or_model)
 181         tags = self.tag_model.get_tag_list(tags)
 182         if not tags:
 183             return queryset
 184         # TODO: presumes reverse generic relation
 185         return queryset.filter(tag_relations__tag__in=tags)
 186
 187     def get_related(self, obj, queryset_or_model):
 188         """
 189         Retrieve a list of instances of the specified model which share
 190         tags with the model instance ``obj``, ordered by the number of
 191         shared tags in descending order.
 192         """
 193         queryset, model = get_queryset_and_model(queryset_or_model)
 194         # TODO: presumes reverse generic relation.
 195         # Do we know it's 'tags'?
 196         return queryset.filter(tag_relations__tag__in=obj.tags).annotate(
 197             count=models.Count('pk')).order_by('-count').exclude(obj=obj.pk)
 198
 199
 200 ##########
 201 # Models #
 202 ##########
 203 def create_intermediary_table_model(model):
 204     """Create an intermediary table model for the specific tag model"""
 205     name = model.__name__ + 'Relation'
 206
 207     class Meta:
 208         db_table = '%s_relation' % model._meta.db_table
 209         unique_together = (('tag', 'content_type', 'object_id'),)
 210         app_label = model._meta.app_label
 211
 212     def obj_unicode(self):
 213         try:
 214             return u'%s [%s]' % (self.content_type.get_object_for_this_type(pk=self.object_id), self.tag)
 215         except ObjectDoesNotExist:
 216             return u'<deleted> [%s]' % self.tag
 217
 218     # Set up a dictionary to simulate declarations within a class
 219     attrs = {
 220         '__module__': model.__module__,
 221         'Meta': Meta,
 222         'tag': models.ForeignKey(model, verbose_name=_('tag'), related_name='items'),
 223         'content_type': models.ForeignKey(ContentType, verbose_name=_('content type')),
 224         'object_id': models.PositiveIntegerField(_('object id'), db_index=True),
 225         'content_object': generic.GenericForeignKey('content_type', 'object_id'),
 226         '__unicode__': obj_unicode,
 227     }
 228
 229     return type(name, (models.Model,), attrs)
 230
 231
 232 class TagMeta(ModelBase):
 233     "Metaclass for tag models (models inheriting from TagBase)."
 234     def __new__(cls, name, bases, attrs):
 235         model = super(TagMeta, cls).__new__(cls, name, bases, attrs)
 236         if not model._meta.abstract:
 237             # Create an intermediary table and register custom managers for concrete models
 238             model.intermediary_table_model = create_intermediary_table_model(model)
 239             TagManager(model.intermediary_table_model).contribute_to_class(model, 'objects')
 240             TaggedItemManager(model).contribute_to_class(model.intermediary_table_model, 'objects')
 241         return model
 242
 243
 244 class TagBase(models.Model):
 245     """Abstract class to be inherited by model classes."""
 246     __metaclass__ = TagMeta
 247
 248     class Meta:
 249         abstract = True
 250
 251     @staticmethod
 252     def get_tag_list(tag_list):
 253         """
 254         Utility function for accepting tag input in a flexible manner.
 255
 256         You should probably override this method in your subclass.
 257         """
 258         if isinstance(tag_list, TagBase):
 259             return [tag_list]
 260         else:
 261             return tag_list
 262