Skip to content

Allow tagman to return model querysets unevaluated #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jamescw wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Allow tagman to return model querysets unevaluated #15

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 95 additions & 54 deletions src/tagman/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,26 @@ def get_query_set(self):
By default return only those objects that are not flagged as
'system' tags.
"""
return super(TagManager, self).get_query_set()\
.exclude(group__system=not self.system_tags).filter(archived=self.archived)
return super(TagManager, self)\
.get_query_set()\
.exclude(group__system=not self.system_tags)\
.filter(archived=self.archived)

def get_tags_with_weight(self, ignore_models=[], composite_name=True):
"""
:param ignore_models: Models to ignore in Tag usage results.
:param composite_name: If True the group name will be prepended
to tag name for dict keys.

Returns dictionary of tag name as key and tag weight (usage) as
value.
"""
tag_dict = {}
tags = super(TagManager, self).get_query_set()
for tag in tags:
tag_name = tag.__unicode__() if composite_name else tag.name
tag_dict[tag_name] = tag.tag_weight(ignore_models)
return tag_dict


class Tag(models.Model):
Expand All @@ -78,8 +96,8 @@ class Tag(models.Model):
archived = models.BooleanField(default=False)

objects = models.Manager()
sys_objects = TagManager(sys=True)
public_objects = TagManager(sys=False)
sys_objects = TagManager(sys=True, archived=False)
public_objects = TagManager(sys=False, archived=False)
non_archived = TagManager(sys=False, archived=False)

def save(self, *args, **kwargs):
Expand Down Expand Up @@ -107,7 +125,7 @@ def archive(self):
"""
Set the archive flag to implement soft-delete
"""
self.archived=True
self.archived = True
self.save()

@property
Expand All @@ -120,61 +138,50 @@ def models_for_tag(self):
instances tagged with this tag.
@todo: This is *really* hacky. Can we do it more elegantly?
"""
models = []
models = set()
for attribute in dir(self):
if attribute[-4:] == '_set':
# we just want the model name, not the set name
model_name = attribute.split('_')[0]
#TODO: check if the list already contains the model name?
models.append(model_name)
# return the unique set of model names
#TODO: what is more efficent?
return set(models)
models.add(model_name)
# return the unique set of model names
return models

def tagged_model_items(self, model_cls=None, model_name="", limit=None,
only_auto=False, filter_dict=None):
def tagged_model_items(self, model_cls=None, model_name="",
only_auto=False):
"""
Return a unique set of instances of a given model, the class for
Return a query_set of a given model, the class for
which is passed into model_cls OR the name for which is passed in
model_name, that are tagged with this tag.

If `only_auto`==True then return only auto-tagged instances.
If `only_auto`==True then return only auto-tagged sets.
"""
def _get_models_items(query_set):
items = []
def _get_model_query_set(set_name):
query_set = None
try:
_set = getattr(self, query_set)
_set = getattr(self, set_name)
except AttributeError:
logger.exception("Set {0} not found on tag {1}".format(
query_set,
set_name,
self
))
else:
if filter_dict:
try:
items = _set.filter(**filter_dict)[:limit]
except FieldError, e:
logger.exception(
"Cannot apply filter {0} to set {1}"
.format(
filter_dict,
query_set
)
)
else:
items = _set.all()[:limit]
return set(items)
query_set = _set
return query_set

if model_cls:
cls_name = model_cls.__name__.lower()
else:
cls_name = model_name.lower()

model_set = set()
if not only_auto:
model_set.update(_get_models_items("{0}_set".format(cls_name)))
model_set.update(_get_models_items("{0}_auto_tagged_set"
.format(cls_name)))
model_set = _get_model_query_set(
"{0}_set".format(cls_name)
)
else:
model_set = _get_model_query_set(
"{0}_auto_tagged_set".format(cls_name)
)

return model_set

Expand All @@ -185,38 +192,72 @@ def auto_tagged_model_items(self, model_cls=None, model_name="",
and tag. See tagged_model_items which this calls with
only_auto=True
"""
return self.tagged_model_items(model_cls, model_name, limit,
only_auto=True)
return self.tagged_model_items(model_cls, model_name, only_auto=True)

def tagged_items(self, limit=None, only_auto=False, ignore_models=[],
filter_dict=None):
def tagged_items(self, only_auto=False, models=None, ignore_models=None):
"""
Return a dictionary, keyed on model name, with each value the
set of items of that model tagged with this tag.
query_set of items of that model tagged with this tag.

:param models:
A list of model classes for which to retrieve items. If absent,
retrieve any model that has a foreign key to a tag.
:param ignore_models:
Model classes not to include in the list of retrieved items.
"""
models = self.models_for_tag()
ignore_models = [model.lower() for model in ignore_models]
if models is None:
models = self.models_for_tag()
else:
models = [model.__name__.lower() for model in models]

if ignore_models is None:
ignore_models = set()
else:
ignore_models = set([model.__name__.lower()
for model in ignore_models])

rdict = {}
for model in models:
if model not in ignore_models:
rdict[model] = self.tagged_model_items(model_name=model,
only_auto=only_auto,
filter_dict=filter_dict,
limit=limit)
only_auto=only_auto)
return rdict

def unique_item_set(self, limit=None, only_auto=False, ignore_models=[],
filter_dict=None):
def tag_weight(self, ignore_models=[]):
"""
Returns the weight of a tag based on the tags usage.
"""
Return the unique item set for a tag
weight = 0
for model_set in self.tagged_items(
ignore_models=ignore_models
).values():
weight += model_set.count()
return weight

def unique_item_set(self, limit=None, only_auto=False, models=None,
ignore_models=None, filter_dict=None):
"""
Return the unique item set for a tag.

:param models:
Return only instances from these models. If absent, retrieve any
model that has a foreign key to Tag
:param ignore_models:
Do not retrieve instances of these models
"""
item_set = set()
tagged_items = self.tagged_items(limit=limit,
only_auto=only_auto,
filter_dict=filter_dict,
tagged_items = self.tagged_items(only_auto=only_auto,
models=models,
ignore_models=ignore_models)
# merge all tagged items into a unique set
item_set.update(itertools.chain(*tagged_items.values()))
for model_set in tagged_items.values():
if model_set:
model_set = model_set.filter(**filter_dict) \
if filter_dict else model_set.all()
item_set.update(
# query set evaluated here
model_set[:limit]
)

return item_set

Expand Down
17 changes: 10 additions & 7 deletions src/tagman/tests/test_models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ def test_get_tagged_model_items(self):
self.item.tags.add(self.tag1)
model_items =\
self.tag1.tagged_model_items(model_cls=self.item.__class__)
self.assertTrue(self.item in model_items)
self.assertTrue(self.item in list(model_items.all()))

def test_get_tag_for_string(self):
self.group.tag_set.add(self.tag1)
Expand Down Expand Up @@ -98,11 +98,11 @@ def test_get_tagged_items(self):
ignored_model_name = ignored_model.__class__.__name__.lower()
# first check the default call returns both
items = self.tag1.tagged_items()
self.assertTrue(self.item in items[item_model_name])
self.assertTrue(ignored_model in items[ignored_model_name])
self.assertTrue(self.item in list(items[item_model_name].all()))
self.assertTrue(ignored_model in list(items[ignored_model_name].all()))
# now ignore the model and check that its not returned
items = self.tag1.tagged_items(ignore_models=[ignored_model_name])
self.assertTrue(self.item in items[item_model_name])
items = self.tag1.tagged_items(ignore_models=[IgnoreTestItem])
self.assertTrue(self.item in list(items[item_model_name].all()))
self.assertTrue(ignored_model_name not in items.keys())

def test_get_unique_item_set(self):
Expand Down Expand Up @@ -217,7 +217,10 @@ def test_auto_tag_creation(self):
def test_get_auto_tagged_items(self):
self.tci.associate_auto_tags()
auto_tag = self.tci.auto_tags.all()[0]
tci_models = auto_tag.tagged_model_items(model_cls=self.tci.__class__)
tci_models = list(
auto_tag.auto_tagged_model_items
(model_cls=self.tci.__class__).all()
)
self.assertTrue(self.tci in tci_models)

def test_get_tci_from_auto_tag(self):
Expand All @@ -226,6 +229,6 @@ def test_get_tci_from_auto_tag(self):
our TCI based on its auto-tag.
"""
auto_tag = Tag.objects.get(slug='tci-slug')
item_set = auto_tag.auto_tagged_model_items(model_cls=TCI)
item_set = set(auto_tag.auto_tagged_model_items(model_cls=TCI).all())
self.assertEquals(len(item_set), 1)
self.assertEquals(item_set.pop(), self.tci)