src/newtagging/models.py

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