<s> from django . conf import settings <EOL> from django . db import connection , models <EOL> try : <EOL> from django . db . models . sql . aggregates import Aggregate <EOL> except ImportError : <EOL> supports_aggregates = False <EOL> else : <EOL> supports_aggregates = True <EOL> from django . contrib . contenttypes . models import ContentType <EOL> if supports_aggregates : <EOL> class CoalesceWrapper ( Aggregate ) : <EOL> sql_template = '<STR_LIT>' <EOL> def __init__ ( self , lookup , ** extra ) : <EOL> self . lookup = lookup <EOL> self . extra = extra <EOL> def _default_alias ( self ) : <EOL> return '<STR_LIT>' % ( self . lookup , self . __class__ . __name__ . lower ( ) ) <EOL> default_alias = property ( _default_alias ) <EOL> def add_to_query ( self , query , alias , col , source , is_summary ) : <EOL> super ( CoalesceWrapper , self ) . __init__ ( col , source , is_summary , ** self . extra ) <EOL> query . aggregate_select [ alias ] = self <EOL> class CoalesceSum ( CoalesceWrapper ) : <EOL> sql_function = '<STR_LIT>' <EOL> class CoalesceCount ( CoalesceWrapper ) : <EOL> sql_function = '<STR_LIT>' <EOL> class VoteManager ( models . Manager ) : <EOL> def get_score ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> ctype = ContentType . objects . get_for_model ( obj ) <EOL> result = self . filter ( object_id = obj . _get_pk_val ( ) , <EOL> content_type = ctype ) . extra ( <EOL> select = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) . values_list ( '<STR_LIT>' , '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> return { <EOL> '<STR_LIT>' : int ( result [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' : int ( result [ <NUM_LIT:1> ] ) , <EOL> } <EOL> def get_scores_in_bulk ( self , objects ) : <EOL> """<STR_LIT>""" <EOL> object_ids = [ o . _get_pk_val ( ) for o in objects ] <EOL> if not object_ids : <EOL> return { } <EOL> ctype = ContentType . objects . get_for_model ( objects [ <NUM_LIT:0> ] ) <EOL> ctype = ContentType . objects . get_for_model ( objects [ <NUM_LIT:0> ] ) <EOL> if supports_aggregates : <EOL> queryset = self . filter ( <EOL> object_id__in = object_ids , <EOL> content_type = ctype , <EOL> ) . values ( <EOL> '<STR_LIT>' , <EOL> ) . annotate ( <EOL> score = CoalesceSum ( '<STR_LIT>' , default = '<STR_LIT:0>' ) , <EOL> num_votes = CoalesceCount ( '<STR_LIT>' , default = '<STR_LIT:0>' ) , <EOL> ) <EOL> else : <EOL> queryset = self . filter ( <EOL> object_id__in = object_ids , <EOL> content_type = ctype , <EOL> ) . extra ( <EOL> select = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> ) . values ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> queryset . query . group_by . append ( '<STR_LIT>' ) <EOL> vote_dict = { } <EOL> for row in queryset : <EOL> vote_dict [ row [ '<STR_LIT>' ] ] = { <EOL> '<STR_LIT>' : int ( row [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : int ( row [ '<STR_LIT>' ] ) , <EOL> } <EOL> return vote_dict <EOL> def record_vote ( self , obj , user , vote ) : <EOL> """<STR_LIT>""" <EOL> if vote not in ( + <NUM_LIT:1> , <NUM_LIT:0> , - <NUM_LIT:1> ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> ctype = ContentType . objects . get_for_model ( obj ) <EOL> try : <EOL> v = self . get ( user = user , content_type = ctype , <EOL> object_id = obj . _get_pk_val ( ) ) <EOL> if v . vote == <NUM_LIT:1> : <EOL> obj . yeas -= <NUM_LIT:1> <EOL> if v . vote == - <NUM_LIT:1> : <EOL> obj . nays -= <NUM_LIT:1> <EOL> if vote == <NUM_LIT:0> : <EOL> v . delete ( ) <EOL> obj . votes -= <NUM_LIT:1> <EOL> else : <EOL> v . vote = vote <EOL> v . save ( ) <EOL> except models . ObjectDoesNotExist : <EOL> if vote != <NUM_LIT:0> : <EOL> self . create ( user = user , content_type = ctype , <EOL> object_id = obj . _get_pk_val ( ) , vote = vote ) <EOL> obj . votes += <NUM_LIT:1> <EOL> if vote == <NUM_LIT:1> : <EOL> obj . yeas += <NUM_LIT:1> <EOL> if vote == - <NUM_LIT:1> : <EOL> obj . nays += <NUM_LIT:1> <EOL> obj . save ( ) <EOL> def get_top ( self , Model , limit = <NUM_LIT:10> , reversed = False ) : <EOL> """<STR_LIT>""" <EOL> ctype = ContentType . objects . get_for_model ( Model ) <EOL> query = """<STR_LIT>""" % ( <EOL> connection . ops . quote_name ( '<STR_LIT>' ) , <EOL> connection . ops . quote_name ( self . model . _meta . db_table ) , <EOL> ) <EOL> if settings . DATABASE_ENGINE == '<STR_LIT>' : <EOL> having_score = connection . ops . quote_name ( '<STR_LIT>' ) <EOL> else : <EOL> having_score = '<STR_LIT>' <EOL> if reversed : <EOL> having_sql = '<STR_LIT>' <EOL> else : <EOL> having_sql = '<STR_LIT>' <EOL> query += having_sql % { <EOL> '<STR_LIT>' : having_score , <EOL> } <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( query , [ ctype . id , limit ] ) <EOL> results = cursor . fetchall ( ) <EOL> objects = Model . objects . in_bulk ( [ id for id , score in results ] ) <EOL> for id , score in results : <EOL> if id in objects : <EOL> yield objects [ id ] , int ( score ) <EOL> def get_bottom ( self , Model , limit = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> return self . get_top ( Model , limit , True ) <EOL> def get_for_user ( self , obj , user ) : <EOL> """<STR_LIT>""" <EOL> if not user . is_authenticated ( ) : <EOL> return None <EOL> ctype = ContentType . objects . get_for_model ( obj ) <EOL> try : <EOL> vote = self . get ( content_type = ctype , object_id = obj . _get_pk_val ( ) , <EOL> user = user ) <EOL> except models . ObjectDoesNotExist : <EOL> vote = None <EOL> return vote <EOL> def get_for_user_in_bulk ( self , objects , user ) : <EOL> """<STR_LIT>""" <EOL> vote_dict = { } <EOL> if len ( objects ) > <NUM_LIT:0> : <EOL> ctype = ContentType . objects . get_for_model ( objects [ <NUM_LIT:0> ] ) <EOL> votes = list ( self . filter ( content_type__pk = ctype . id , <EOL> object_id__in = [ obj . _get_pk_val ( ) for obj in objects ] , <EOL> user__pk = user . id ) ) <EOL> vote_dict = dict ( [ ( vote . object_id , vote ) for vote in votes ] ) <EOL> return vote_dict </s>
