My favorites | Sign in
Project Home Wiki Issues Source
Checkout   Browse   Changes  
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
from django.conf import settings
from django.db import connection, models
from django.contrib.contenttypes.models import ContentType

qn = connection.ops.quote_name

class VoteManager(models.Manager):
def get_score(self, obj):
"""
Get a dictionary containing the total score for ``obj`` and
the number of votes it's received.
"""
query = """
SELECT SUM(vote), COUNT(vote)
FROM %s
WHERE content_type_id = %%s
AND object_id = %%s""" % qn(self.model._meta.db_table)
ctype = ContentType.objects.get_for_model(obj)
cursor = connection.cursor()
cursor.execute(query, [ctype.id, obj._get_pk_val()])
result = cursor.fetchall()[0]
# MySQL returns floats and longs respectively for these
# results, so we need to convert them to ints explicitly.
return {
'score': result[0] and int(result[0]) or 0,
'num_votes': int(result[1]),
}

def get_scores_in_bulk(self, objects):
"""
Get a dictionary mapping object ids to total score and number
of votes for each object.
"""
query = """
SELECT object_id, SUM(vote), COUNT(vote)
FROM %s
WHERE content_type_id = %%s
AND object_id IN (%s)
GROUP BY object_id""" % (
qn(self.model._meta.db_table),
','.join(['%s'] * len(objects))
)
ctype = ContentType.objects.get_for_model(objects[0])
cursor = connection.cursor()
cursor.execute(query, [ctype.id] + [obj._get_pk_val() \
for obj in objects])
results = cursor.fetchall()
return dict([(int(object_id), {
'score': int(score),
'num_votes': int(num_votes),
}) for object_id, score, num_votes in results])

def record_vote(self, obj, user, vote):
"""
Record a user's vote on a given object. Only allows a given user
to vote once, though that vote may be changed.

A zero vote indicates that any existing vote should be removed.
"""
if vote not in (+1, 0, -1):
raise ValueError('Invalid vote (must be +1/0/-1)')
ctype = ContentType.objects.get_for_model(obj)
try:
v = self.get(user=user, content_type=ctype,
object_id=obj._get_pk_val())
if vote == 0:
v.delete()
else:
v.vote = vote
v.save()
except models.ObjectDoesNotExist:
if vote != 0:
self.create(user=user, content_type=ctype,
object_id=obj._get_pk_val(), vote=vote)

def get_top(self, Model, limit=10, reversed=False):
"""
Get the top N scored objects for a given model.

Yields (object, score) tuples.
"""
ctype = ContentType.objects.get_for_model(Model)
query = """
SELECT object_id, SUM(vote) as %s
FROM %s
WHERE content_type_id = %%s
GROUP BY object_id""" % (
qn('score'),
qn(self.model._meta.db_table),
)

# MySQL has issues with re-using the aggregate function in the
# HAVING clause, so we alias the score and use this alias for
# its benefit.
if settings.DATABASE_ENGINE == 'mysql':
having_score = qn('score')
else:
having_score = 'SUM(vote)'
if reversed:
having_sql = ' HAVING %(having_score)s < 0 ORDER BY %(having_score)s ASC %(limit_offset)s'
else:
having_sql = ' HAVING %(having_score)s > 0 ORDER BY %(having_score)s DESC %(limit_offset)s'
query += having_sql % {
'having_score': having_score,
'limit_offset': connection.ops.limit_offset_sql(limit),
}

cursor = connection.cursor()
cursor.execute(query, [ctype.id])
results = cursor.fetchall()

# Use in_bulk() to avoid O(limit) db hits.
objects = Model.objects.in_bulk([id for id, score in results])

# Yield each object, score pair. Because of the lazy nature of generic
# relations, missing objects are silently ignored.
for id, score in results:
if id in objects:
yield objects[id], int(score)

def get_bottom(self, Model, limit=10):
"""
Get the bottom (i.e. most negative) N scored objects for a given
model.

Yields (object, score) tuples.
"""
return self.get_top(Model, limit, True)

def get_for_user(self, obj, user):
"""
Get the vote made on the given object by the given user, or
``None`` if no matching vote exists.
"""
if not user.is_authenticated():
return None
ctype = ContentType.objects.get_for_model(obj)
try:
vote = self.get(content_type=ctype, object_id=obj._get_pk_val(),
user=user)
except models.ObjectDoesNotExist:
vote = None
return vote

def get_for_user_in_bulk(self, objects, user):
"""
Get a dictionary mapping object ids to votes made by the given
user on the corresponding objects.
"""
vote_dict = {}
if len(objects) > 0:
ctype = ContentType.objects.get_for_model(objects[0])
votes = list(self.filter(content_type__pk=ctype.id,
object_id__in=[obj._get_pk_val() \
for obj in objects],
user__pk=user.id))
vote_dict = dict([(vote.object_id, vote) for vote in votes])
return vote_dict

Change log

r56 by jonathan.buchanan on Aug 20, 2007   Diff
Brought up to date with latest Django SVN
version so tests run again
Go to: 
Project members, sign in to write a code review

Older revisions

r55 by jonathan.buchanan on Jun 1, 2007   Diff
Replaced a COUNT(*)
r54 by jonathan.buchanan on Jun 1, 2007   Diff
Fix for get_top query in MySQL; all
tests now pass in MySQL
r53 by jonathan.buchanan on May 31, 2007   Diff
Moved VoteManager to voting.managers
All revisions of this file

File info

Size: 6017 bytes, 158 lines
Powered by Google Project Hosting