from django.db import connection, models
from django.utils.translation import ugettext_lazy as _
from django.db.models.base import ModelBase
from django.db import connection, models
from django.utils.translation import ugettext_lazy as _
from django.db.models.base import ModelBase
def get_queryset_and_model(queryset_or_model):
"""
Given a ``QuerySet`` or a ``Model``, returns a two-tuple of
def get_queryset_and_model(queryset_or_model):
"""
Given a ``QuerySet`` or a ``Model``, returns a two-tuple of
def __init__(self, intermediary_table_model):
super(TagManager, self).__init__()
self.intermediary_table_model = intermediary_table_model
def __init__(self, intermediary_table_model):
super(TagManager, self).__init__()
self.intermediary_table_model = intermediary_table_model
+ models.signals.pre_delete.connect(self.target_deleted)
+
+ def target_deleted(self, instance, **kwargs):
+ """ clear tag relations before deleting an object """
+ try:
+ int(instance.pk)
+ except ValueError:
+ return
+
+ self.update_tags(instance, [])
+
def update_tags(self, obj, tags):
"""
Update tags associated with an object.
def update_tags(self, obj, tags):
"""
Update tags associated with an object.
current_tags = list(self.filter(items__content_type__pk=content_type.pk,
items__object_id=obj.pk))
updated_tags = self.model.get_tag_list(tags)
current_tags = list(self.filter(items__content_type__pk=content_type.pk,
items__object_id=obj.pk))
updated_tags = self.model.get_tag_list(tags)
# Remove tags which no longer apply
tags_for_removal = [tag for tag in current_tags \
if tag not in updated_tags]
# Remove tags which no longer apply
tags_for_removal = [tag for tag in current_tags \
if tag not in updated_tags]
if tag not in current_tags:
self.intermediary_table_model._default_manager.create(tag=tag, content_object=obj)
if tag not in current_tags:
self.intermediary_table_model._default_manager.create(tag=tag, content_object=obj)
+
+ tags_updated.send(sender=obj, affected_tags=tags_to_add + tags_for_removal)
+
+ def remove_tag(self, obj, tag):
+ """
+ Remove tag from an object.
+ """
+ content_type = ContentType.objects.get_for_model(obj)
+ self.intermediary_table_model._default_manager.filter(content_type__pk=content_type.pk,
+ object_id=obj.pk, tag=tag).delete()
+
def get_for_object(self, obj):
"""
Create a queryset matching all tags associated with the given
def get_for_object(self, obj):
"""
Create a queryset matching all tags associated with the given
ctype = ContentType.objects.get_for_model(obj)
return self.filter(items__content_type__pk=ctype.pk,
items__object_id=obj.pk)
ctype = ContentType.objects.get_for_model(obj)
return self.filter(items__content_type__pk=ctype.pk,
items__object_id=obj.pk)
def _get_usage(self, model, counts=False, min_count=None, extra_joins=None, extra_criteria=None, params=None, extra=None):
"""
Perform the custom SQL query for ``usage_for_model`` and
def _get_usage(self, model, counts=False, min_count=None, extra_joins=None, extra_criteria=None, params=None, extra=None):
"""
Perform the custom SQL query for ``usage_for_model`` and
model_table = qn(model._meta.db_table)
model_pk = '%s.%s' % (model_table, qn(model._meta.pk.column))
tag_columns = self._get_tag_columns()
model_table = qn(model._meta.db_table)
model_pk = '%s.%s' % (model_table, qn(model._meta.pk.column))
tag_columns = self._get_tag_columns()
%%s
ORDER BY %(tag)s.%(ordering)s ASC""" % {
'tag': qn(self.model._meta.db_table),
%%s
ORDER BY %(tag)s.%(ordering)s ASC""" % {
'tag': qn(self.model._meta.db_table),
- extra_joins = ' '.join(queryset.query.get_from_clause()[0][1:])
- where, params = queryset.query.where.as_sql()
+ if getattr(queryset.query, 'get_compiler', None):
+ # Django 1.2+
+ compiler = queryset.query.get_compiler(using='default')
+ extra_joins = ' '.join(compiler.get_from_clause()[0][1:])
+ where, params = queryset.query.where.as_sql(
+ compiler.quote_name_unless_alias, compiler.connection
+ )
+ else:
+ # Django pre-1.2
+ extra_joins = ' '.join(queryset.query.get_from_clause()[0][1:])
+ where, params = queryset.query.where.as_sql()
+
tag_count = len(tags)
tagged_item_table = qn(self.intermediary_table_model._meta.db_table)
tag_columns = self._get_tag_columns()
tag_count = len(tags)
tagged_item_table = qn(self.intermediary_table_model._meta.db_table)
tag_columns = self._get_tag_columns()
query = """
SELECT %(tag_columns)s%(count_sql)s
FROM %(tagged_item)s INNER JOIN %(tag)s ON %(tagged_item)s.tag_id = %(tag)s.id
WHERE %(tagged_item)s.content_type_id = %(content_type_id)s
query = """
SELECT %(tag_columns)s%(count_sql)s
FROM %(tagged_item)s INNER JOIN %(tag)s ON %(tagged_item)s.tag_id = %(tag)s.id
WHERE %(tagged_item)s.content_type_id = %(content_type_id)s
- AND %(tagged_item)s.object_id IN
- (
- SELECT %(tagged_item)s.object_id
- FROM %(tagged_item)s, %(tag)s
- WHERE %(tagged_item)s.content_type_id = %(content_type_id)s
- AND %(tag)s.id = %(tagged_item)s.tag_id
- AND %(tag)s.id IN (%(tag_id_placeholders)s)
- GROUP BY %(tagged_item)s.object_id
- HAVING COUNT(%(tagged_item)s.object_id) = %(tag_count)s
- )
- AND %(tag)s.id NOT IN (%(tag_id_placeholders)s)
- %(extra_where)s
+ AND %(tagged_item)s.object_id IN
+ (
+ SELECT *
+ FROM (
+ SELECT %(tagged_item)s.object_id
+ FROM %(tagged_item)s, %(tag)s
+ WHERE %(tagged_item)s.content_type_id = %(content_type_id)s
+ AND %(tag)s.id = %(tagged_item)s.tag_id
+ AND %(tag)s.id IN (%(tag_id_placeholders)s)
+ GROUP BY %(tagged_item)s.object_id
+ HAVING COUNT(%(tagged_item)s.object_id) = %(tag_count)s
+ ) AS temporary
+ )
+ AND %(tag)s.id NOT IN (%(tag_id_placeholders)s)
+ %(extra_where)s
def __init__(self, tag_model):
super(TaggedItemManager, self).__init__()
self.tag_model = tag_model
def __init__(self, tag_model):
super(TaggedItemManager, self).__init__()
self.tag_model = tag_model
def get_by_model(self, queryset_or_model, tags):
"""
Create a ``QuerySet`` containing instances of the specified
def get_by_model(self, queryset_or_model, tags):
"""
Create a ``QuerySet`` containing instances of the specified
- def get_related(self, obj, queryset_or_model, num=None):
+ def get_related(self, obj, queryset_or_model, num=None, ignore_by_tag=None):
"""
Retrieve a list of instances of the specified model which share
tags with the model instance ``obj``, ordered by the number of
"""
Retrieve a list of instances of the specified model which share
tags with the model instance ``obj``, ordered by the number of
"""
queryset, model = get_queryset_and_model(queryset_or_model)
model_table = qn(model._meta.db_table)
"""
queryset, model = get_queryset_and_model(queryset_or_model)
model_table = qn(model._meta.db_table)
+ if ignore_by_tag is not None:
+ query += """
+ AND NOT EXISTS (
+ SELECT * FROM %(tagged_item)s
+ WHERE %(tagged_item)s.object_id = %(model_pk)s
+ AND %(tagged_item)s.content_type_id = %(content_type_id)s
+ AND %(ignore_id)s = %(tagged_item)s.tag_id
+ )
+ """
'content_type_id': content_type.pk,
'related_content_type_id': related_content_type.pk,
'limit_offset': num is not None and connection.ops.limit_offset_sql(num) or '',
'content_type_id': content_type.pk,
'related_content_type_id': related_content_type.pk,
'limit_offset': num is not None and connection.ops.limit_offset_sql(num) or '',
def create_intermediary_table_model(model):
"""Create an intermediary table model for the specific tag model"""
name = model.__name__ + 'Relation'
def create_intermediary_table_model(model):
"""Create an intermediary table model for the specific tag model"""
name = model.__name__ + 'Relation'
class Meta:
db_table = '%s_relation' % model._meta.db_table
unique_together = (('tag', 'content_type', 'object_id'),)
def obj_unicode(self):
class Meta:
db_table = '%s_relation' % model._meta.db_table
unique_together = (('tag', 'content_type', 'object_id'),)
def obj_unicode(self):
- return u'%s [%s]' % (self.content_type.get_object_for_this_type(pk=self.object_id), self.tag)
-
- # Set up a dictionary to simulate declarations within a class
+ try:
+ return u'%s [%s]' % (self.content_type.get_object_for_this_type(pk=self.object_id), self.tag)
+ except ObjectDoesNotExist:
+ return u'<deleted> [%s]' % self.tag
+
+ # Set up a dictionary to simulate declarations within a class
You should probably override this method in your subclass.
"""
if isinstance(tag_list, TagBase):
You should probably override this method in your subclass.
"""
if isinstance(tag_list, TagBase):