<s> from __future__ import unicode_literals <EOL> from collections import defaultdict <EOL> from django . contrib . contenttypes . models import ContentType <EOL> from django . core import checks <EOL> from django . core . exceptions import FieldDoesNotExist , ObjectDoesNotExist <EOL> from django . db import DEFAULT_DB_ALIAS , models , router , transaction <EOL> from django . db . models import DO_NOTHING , signals <EOL> from django . db . models . base import ModelBase , make_foreign_order_accessors <EOL> from django . db . models . fields . related import ( <EOL> ForeignObject , ForeignObjectRel , ReverseManyToOneDescriptor , <EOL> lazy_related_operation , <EOL> ) <EOL> from django . db . models . query_utils import PathInfo <EOL> from django . utils . encoding import python_2_unicode_compatible , smart_text <EOL> from django . utils . functional import cached_property <EOL> @ python_2_unicode_compatible <EOL> class GenericForeignKey ( object ) : <EOL> """<STR_LIT>""" <EOL> auto_created = False <EOL> concrete = False <EOL> editable = False <EOL> hidden = False <EOL> is_relation = True <EOL> many_to_many = False <EOL> many_to_one = True <EOL> one_to_many = False <EOL> one_to_one = False <EOL> related_model = None <EOL> remote_field = None <EOL> def __init__ ( self , ct_field = '<STR_LIT>' , fk_field = '<STR_LIT>' , for_concrete_model = True ) : <EOL> self . ct_field = ct_field <EOL> self . fk_field = fk_field <EOL> self . for_concrete_model = for_concrete_model <EOL> self . editable = False <EOL> self . rel = None <EOL> self . column = None <EOL> def contribute_to_class ( self , cls , name , ** kwargs ) : <EOL> self . name = name <EOL> self . model = cls <EOL> self . cache_attr = "<STR_LIT>" % name <EOL> cls . _meta . add_field ( self , private = True ) <EOL> if not cls . _meta . abstract : <EOL> signals . pre_init . connect ( self . instance_pre_init , sender = cls ) <EOL> setattr ( cls , name , self ) <EOL> def get_filter_kwargs_for_object ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> self . fk_field : getattr ( obj , self . fk_field ) , <EOL> self . ct_field : getattr ( obj , self . ct_field ) , <EOL> } <EOL> def get_forward_related_filter ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> self . fk_field : obj . pk , <EOL> self . ct_field : ContentType . objects . get_for_model ( obj ) . pk , <EOL> } <EOL> def __str__ ( self ) : <EOL> model = self . model <EOL> app = model . _meta . app_label <EOL> return '<STR_LIT>' % ( app , model . _meta . object_name , self . name ) <EOL> def check ( self , ** kwargs ) : <EOL> errors = [ ] <EOL> errors . extend ( self . _check_field_name ( ) ) <EOL> errors . extend ( self . _check_object_id_field ( ) ) <EOL> errors . extend ( self . _check_content_type_field ( ) ) <EOL> return errors <EOL> def _check_field_name ( self ) : <EOL> if self . name . endswith ( "<STR_LIT:_>" ) : <EOL> return [ <EOL> checks . Error ( <EOL> '<STR_LIT>' , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> else : <EOL> return [ ] <EOL> def _check_object_id_field ( self ) : <EOL> try : <EOL> self . model . _meta . get_field ( self . fk_field ) <EOL> except FieldDoesNotExist : <EOL> return [ <EOL> checks . Error ( <EOL> "<STR_LIT>" % self . fk_field , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> else : <EOL> return [ ] <EOL> def _check_content_type_field ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> field = self . model . _meta . get_field ( self . ct_field ) <EOL> except FieldDoesNotExist : <EOL> return [ <EOL> checks . Error ( <EOL> "<STR_LIT>" % ( <EOL> self . model . _meta . object_name , self . ct_field <EOL> ) , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> else : <EOL> if not isinstance ( field , models . ForeignKey ) : <EOL> return [ <EOL> checks . Error ( <EOL> "<STR_LIT>" % ( <EOL> self . model . _meta . object_name , self . ct_field <EOL> ) , <EOL> hint = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> elif field . remote_field . model != ContentType : <EOL> return [ <EOL> checks . Error ( <EOL> "<STR_LIT>" % ( <EOL> self . model . _meta . object_name , self . ct_field <EOL> ) , <EOL> hint = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> else : <EOL> return [ ] <EOL> def instance_pre_init ( self , signal , sender , args , kwargs , ** _kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . name in kwargs : <EOL> value = kwargs . pop ( self . name ) <EOL> if value is not None : <EOL> kwargs [ self . ct_field ] = self . get_content_type ( obj = value ) <EOL> kwargs [ self . fk_field ] = value . _get_pk_val ( ) <EOL> else : <EOL> kwargs [ self . ct_field ] = None <EOL> kwargs [ self . fk_field ] = None <EOL> def get_content_type ( self , obj = None , id = None , using = None ) : <EOL> if obj is not None : <EOL> return ContentType . objects . db_manager ( obj . _state . db ) . get_for_model ( <EOL> obj , for_concrete_model = self . for_concrete_model ) <EOL> elif id is not None : <EOL> return ContentType . objects . db_manager ( using ) . get_for_id ( id ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def get_prefetch_queryset ( self , instances , queryset = None ) : <EOL> if queryset is not None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> fk_dict = defaultdict ( set ) <EOL> instance_dict = { } <EOL> ct_attname = self . model . _meta . get_field ( self . ct_field ) . get_attname ( ) <EOL> for instance in instances : <EOL> ct_id = getattr ( instance , ct_attname ) <EOL> if ct_id is not None : <EOL> fk_val = getattr ( instance , self . fk_field ) <EOL> if fk_val is not None : <EOL> fk_dict [ ct_id ] . add ( fk_val ) <EOL> instance_dict [ ct_id ] = instance <EOL> ret_val = [ ] <EOL> for ct_id , fkeys in fk_dict . items ( ) : <EOL> instance = instance_dict [ ct_id ] <EOL> ct = self . get_content_type ( id = ct_id , using = instance . _state . db ) <EOL> ret_val . extend ( ct . get_all_objects_for_this_type ( pk__in = fkeys ) ) <EOL> def gfk_key ( obj ) : <EOL> ct_id = getattr ( obj , ct_attname ) <EOL> if ct_id is None : <EOL> return None <EOL> else : <EOL> model = self . get_content_type ( id = ct_id , <EOL> using = obj . _state . db ) . model_class ( ) <EOL> return ( model . _meta . pk . get_prep_value ( getattr ( obj , self . fk_field ) ) , <EOL> model ) <EOL> return ( ret_val , <EOL> lambda obj : ( obj . _get_pk_val ( ) , obj . __class__ ) , <EOL> gfk_key , <EOL> True , <EOL> self . cache_attr ) <EOL> def is_cached ( self , instance ) : <EOL> return hasattr ( instance , self . cache_attr ) <EOL> def __get__ ( self , instance , cls = None ) : <EOL> if instance is None : <EOL> return self <EOL> try : <EOL> return getattr ( instance , self . cache_attr ) <EOL> except AttributeError : <EOL> rel_obj = None <EOL> f = self . model . _meta . get_field ( self . ct_field ) <EOL> ct_id = getattr ( instance , f . get_attname ( ) , None ) <EOL> if ct_id is not None : <EOL> ct = self . get_content_type ( id = ct_id , using = instance . _state . db ) <EOL> try : <EOL> rel_obj = ct . get_object_for_this_type ( pk = getattr ( instance , self . fk_field ) ) <EOL> except ObjectDoesNotExist : <EOL> pass <EOL> setattr ( instance , self . cache_attr , rel_obj ) <EOL> return rel_obj <EOL> def __set__ ( self , instance , value ) : <EOL> ct = None <EOL> fk = None <EOL> if value is not None : <EOL> ct = self . get_content_type ( obj = value ) <EOL> fk = value . _get_pk_val ( ) <EOL> setattr ( instance , self . ct_field , ct ) <EOL> setattr ( instance , self . fk_field , fk ) <EOL> setattr ( instance , self . cache_attr , value ) <EOL> class GenericRel ( ForeignObjectRel ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , field , to , related_name = None , related_query_name = None , limit_choices_to = None ) : <EOL> super ( GenericRel , self ) . __init__ ( <EOL> field , to , <EOL> related_name = related_query_name or '<STR_LIT:+>' , <EOL> related_query_name = related_query_name , <EOL> limit_choices_to = limit_choices_to , <EOL> on_delete = DO_NOTHING , <EOL> ) <EOL> class GenericRelation ( ForeignObject ) : <EOL> """<STR_LIT>""" <EOL> auto_created = False <EOL> many_to_many = False <EOL> many_to_one = False <EOL> one_to_many = True <EOL> one_to_one = False <EOL> rel_class = GenericRel <EOL> def __init__ ( self , to , object_id_field = '<STR_LIT>' , content_type_field = '<STR_LIT>' , <EOL> for_concrete_model = True , related_query_name = None , limit_choices_to = None , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = self . rel_class ( <EOL> self , to , <EOL> related_query_name = related_query_name , <EOL> limit_choices_to = limit_choices_to , <EOL> ) <EOL> kwargs [ '<STR_LIT:blank>' ] = True <EOL> kwargs [ '<STR_LIT>' ] = models . CASCADE <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> super ( GenericRelation , self ) . __init__ ( <EOL> to , from_fields = [ object_id_field ] , to_fields = [ ] , ** kwargs ) <EOL> self . object_id_field_name = object_id_field <EOL> self . content_type_field_name = content_type_field <EOL> self . for_concrete_model = for_concrete_model <EOL> def check ( self , ** kwargs ) : <EOL> errors = super ( GenericRelation , self ) . check ( ** kwargs ) <EOL> errors . extend ( self . _check_generic_foreign_key_existence ( ) ) <EOL> return errors <EOL> def _check_generic_foreign_key_existence ( self ) : <EOL> target = self . remote_field . model <EOL> if isinstance ( target , ModelBase ) : <EOL> fields = target . _meta . private_fields <EOL> if any ( isinstance ( field , GenericForeignKey ) and <EOL> field . ct_field == self . content_type_field_name and <EOL> field . fk_field == self . object_id_field_name <EOL> for field in fields ) : <EOL> return [ ] <EOL> else : <EOL> return [ <EOL> checks . Error ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> target . _meta . app_label , target . _meta . object_name <EOL> ) , <EOL> obj = self , <EOL> id = '<STR_LIT>' , <EOL> ) <EOL> ] <EOL> else : <EOL> return [ ] <EOL> def resolve_related_fields ( self ) : <EOL> self . to_fields = [ self . model . _meta . pk . name ] <EOL> return [ ( self . remote_field . model . _meta . get_field ( self . object_id_field_name ) , self . model . _meta . pk ) ] <EOL> def _get_path_info_with_parent ( self ) : <EOL> """<STR_LIT>""" <EOL> path = [ ] <EOL> opts = self . remote_field . model . _meta <EOL> parent_opts = opts . get_field ( self . object_id_field_name ) . model . _meta <EOL> target = parent_opts . pk <EOL> path . append ( PathInfo ( self . model . _meta , parent_opts , ( target , ) , self . remote_field , True , False ) ) <EOL> parent_field_chain = [ ] <EOL> while parent_opts != opts : <EOL> field = opts . get_ancestor_link ( parent_opts . model ) <EOL> parent_field_chain . append ( field ) <EOL> opts = field . remote_field . model . _meta <EOL> parent_field_chain . reverse ( ) <EOL> for field in parent_field_chain : <EOL> path . extend ( field . remote_field . get_path_info ( ) ) <EOL> return path <EOL> def get_path_info ( self ) : <EOL> opts = self . remote_field . model . _meta <EOL> object_id_field = opts . get_field ( self . object_id_field_name ) <EOL> if object_id_field . model != opts . model : <EOL> return self . _get_path_info_with_parent ( ) <EOL> else : <EOL> target = opts . pk <EOL> return [ PathInfo ( self . model . _meta , opts , ( target , ) , self . remote_field , True , False ) ] <EOL> def get_reverse_path_info ( self ) : <EOL> opts = self . model . _meta <EOL> from_opts = self . remote_field . model . _meta <EOL> return [ PathInfo ( from_opts , opts , ( opts . pk , ) , self , not self . unique , False ) ] <EOL> def value_to_string ( self , obj ) : <EOL> qs = getattr ( obj , self . name ) . all ( ) <EOL> return smart_text ( [ instance . _get_pk_val ( ) for instance in qs ] ) <EOL> def contribute_to_class ( self , cls , name , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> super ( GenericRelation , self ) . contribute_to_class ( cls , name , ** kwargs ) <EOL> self . model = cls <EOL> setattr ( cls , self . name , ReverseGenericManyToOneDescriptor ( self . remote_field ) ) <EOL> def matching_gfk ( field ) : <EOL> return ( <EOL> isinstance ( field , GenericForeignKey ) and <EOL> self . content_type_field_name == field . ct_field and <EOL> self . object_id_field_name == field . fk_field <EOL> ) <EOL> def make_generic_foreign_order_accessors ( related_model , model ) : <EOL> if matching_gfk ( model . _meta . order_with_respect_to ) : <EOL> make_foreign_order_accessors ( model , related_model ) <EOL> lazy_related_operation ( make_generic_foreign_order_accessors , self . model , self . remote_field . model ) <EOL> def set_attributes_from_rel ( self ) : <EOL> pass <EOL> def get_internal_type ( self ) : <EOL> return "<STR_LIT>" <EOL> def get_content_type ( self ) : <EOL> """<STR_LIT>""" <EOL> return ContentType . objects . get_for_model ( self . model , <EOL> for_concrete_model = self . for_concrete_model ) <EOL> def get_extra_restriction ( self , where_class , alias , remote_alias ) : <EOL> field = self . remote_field . model . _meta . get_field ( self . content_type_field_name ) <EOL> contenttype_pk = self . get_content_type ( ) . pk <EOL> cond = where_class ( ) <EOL> lookup = field . get_lookup ( '<STR_LIT>' ) ( field . get_col ( remote_alias ) , contenttype_pk ) <EOL> cond . add ( lookup , '<STR_LIT>' ) <EOL> return cond <EOL> def bulk_related_objects ( self , objs , using = DEFAULT_DB_ALIAS ) : <EOL> """<STR_LIT>""" <EOL> return self . remote_field . model . _base_manager . db_manager ( using ) . filter ( ** { <EOL> "<STR_LIT>" % self . content_type_field_name : ContentType . objects . db_manager ( using ) . get_for_model ( <EOL> self . model , for_concrete_model = self . for_concrete_model ) . pk , <EOL> "<STR_LIT>" % self . object_id_field_name : [ obj . pk for obj in objs ] <EOL> } ) <EOL> class ReverseGenericManyToOneDescriptor ( ReverseManyToOneDescriptor ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def related_manager_cls ( self ) : <EOL> return create_generic_related_manager ( <EOL> self . rel . model . _default_manager . __class__ , <EOL> self . rel , <EOL> ) <EOL> def create_generic_related_manager ( superclass , rel ) : <EOL> """<STR_LIT>""" <EOL> class GenericRelatedObjectManager ( superclass ) : <EOL> def __init__ ( self , instance = None ) : <EOL> super ( GenericRelatedObjectManager , self ) . __init__ ( ) <EOL> self . instance = instance <EOL> self . model = rel . model <EOL> content_type = ContentType . objects . db_manager ( instance . _state . db ) . get_for_model ( <EOL> instance , for_concrete_model = rel . field . for_concrete_model ) <EOL> self . content_type = content_type <EOL> self . content_type_field_name = rel . field . content_type_field_name <EOL> self . object_id_field_name = rel . field . object_id_field_name <EOL> self . prefetch_cache_name = rel . field . attname <EOL> self . pk_val = instance . _get_pk_val ( ) <EOL> self . core_filters = { <EOL> '<STR_LIT>' % self . content_type_field_name : content_type . id , <EOL> self . object_id_field_name : self . pk_val , <EOL> } <EOL> def __call__ ( self , ** kwargs ) : <EOL> manager = getattr ( self . model , kwargs . pop ( '<STR_LIT>' ) ) <EOL> manager_class = create_generic_related_manager ( manager . __class__ , rel ) <EOL> return manager_class ( instance = self . instance ) <EOL> do_not_call_in_templates = True <EOL> def __str__ ( self ) : <EOL> return repr ( self ) <EOL> def _apply_rel_filters ( self , queryset ) : <EOL> """<STR_LIT>""" <EOL> db = self . _db or router . db_for_read ( self . model , instance = self . instance ) <EOL> return queryset . using ( db ) . filter ( ** self . core_filters ) <EOL> def get_queryset ( self ) : <EOL> try : <EOL> return self . instance . _prefetched_objects_cache [ self . prefetch_cache_name ] <EOL> except ( AttributeError , KeyError ) : <EOL> queryset = super ( GenericRelatedObjectManager , self ) . get_queryset ( ) <EOL> return self . _apply_rel_filters ( queryset ) <EOL> def get_prefetch_queryset ( self , instances , queryset = None ) : <EOL> if queryset is None : <EOL> queryset = super ( GenericRelatedObjectManager , self ) . get_queryset ( ) <EOL> queryset . _add_hints ( instance = instances [ <NUM_LIT:0> ] ) <EOL> queryset = queryset . using ( queryset . _db or self . _db ) <EOL> query = { <EOL> '<STR_LIT>' % self . content_type_field_name : self . content_type . id , <EOL> '<STR_LIT>' % self . object_id_field_name : set ( obj . _get_pk_val ( ) for obj in instances ) <EOL> } <EOL> object_id_converter = instances [ <NUM_LIT:0> ] . _meta . pk . to_python <EOL> return ( queryset . filter ( ** query ) , <EOL> lambda relobj : object_id_converter ( getattr ( relobj , self . object_id_field_name ) ) , <EOL> lambda obj : obj . _get_pk_val ( ) , <EOL> False , <EOL> self . prefetch_cache_name ) <EOL> def add ( self , * objs , ** kwargs ) : <EOL> bulk = kwargs . pop ( '<STR_LIT>' , True ) <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> def check_and_update_obj ( obj ) : <EOL> if not isinstance ( obj , self . model ) : <EOL> raise TypeError ( "<STR_LIT>" % ( <EOL> self . model . _meta . object_name , obj <EOL> ) ) <EOL> setattr ( obj , self . content_type_field_name , self . content_type ) <EOL> setattr ( obj , self . object_id_field_name , self . pk_val ) <EOL> if bulk : <EOL> pks = [ ] <EOL> for obj in objs : <EOL> if obj . _state . adding or obj . _state . db != db : <EOL> raise ValueError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % obj <EOL> ) <EOL> check_and_update_obj ( obj ) <EOL> pks . append ( obj . pk ) <EOL> self . model . _base_manager . using ( db ) . filter ( pk__in = pks ) . update ( ** { <EOL> self . content_type_field_name : self . content_type , <EOL> self . object_id_field_name : self . pk_val , <EOL> } ) <EOL> else : <EOL> with transaction . atomic ( using = db , savepoint = False ) : <EOL> for obj in objs : <EOL> check_and_update_obj ( obj ) <EOL> obj . save ( ) <EOL> add . alters_data = True <EOL> def remove ( self , * objs , ** kwargs ) : <EOL> if not objs : <EOL> return <EOL> bulk = kwargs . pop ( '<STR_LIT>' , True ) <EOL> self . _clear ( self . filter ( pk__in = [ o . pk for o in objs ] ) , bulk ) <EOL> remove . alters_data = True <EOL> def clear ( self , ** kwargs ) : <EOL> bulk = kwargs . pop ( '<STR_LIT>' , True ) <EOL> self . _clear ( self , bulk ) <EOL> clear . alters_data = True <EOL> def _clear ( self , queryset , bulk ) : <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> queryset = queryset . using ( db ) <EOL> if bulk : <EOL> queryset . delete ( ) <EOL> else : <EOL> with transaction . atomic ( using = db , savepoint = False ) : <EOL> for obj in queryset : <EOL> obj . delete ( ) <EOL> _clear . alters_data = True <EOL> def set ( self , objs , ** kwargs ) : <EOL> objs = tuple ( objs ) <EOL> bulk = kwargs . pop ( '<STR_LIT>' , True ) <EOL> clear = kwargs . pop ( '<STR_LIT>' , False ) <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> with transaction . atomic ( using = db , savepoint = False ) : <EOL> if clear : <EOL> self . clear ( ) <EOL> self . add ( * objs , bulk = bulk ) <EOL> else : <EOL> old_objs = set ( self . using ( db ) . all ( ) ) <EOL> new_objs = [ ] <EOL> for obj in objs : <EOL> if obj in old_objs : <EOL> old_objs . remove ( obj ) <EOL> else : <EOL> new_objs . append ( obj ) <EOL> self . remove ( * old_objs ) <EOL> self . add ( * new_objs , bulk = bulk ) <EOL> set . alters_data = True <EOL> def create ( self , ** kwargs ) : <EOL> kwargs [ self . content_type_field_name ] = self . content_type <EOL> kwargs [ self . object_id_field_name ] = self . pk_val <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> return super ( GenericRelatedObjectManager , self ) . using ( db ) . create ( ** kwargs ) <EOL> create . alters_data = True <EOL> def get_or_create ( self , ** kwargs ) : <EOL> kwargs [ self . content_type_field_name ] = self . content_type <EOL> kwargs [ self . object_id_field_name ] = self . pk_val <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> return super ( GenericRelatedObjectManager , self ) . using ( db ) . get_or_create ( ** kwargs ) <EOL> get_or_create . alters_data = True <EOL> def update_or_create ( self , ** kwargs ) : <EOL> kwargs [ self . content_type_field_name ] = self . content_type <EOL> kwargs [ self . object_id_field_name ] = self . pk_val <EOL> db = router . db_for_write ( self . model , instance = self . instance ) <EOL> return super ( GenericRelatedObjectManager , self ) . using ( db ) . update_or_create ( ** kwargs ) <EOL> update_or_create . alters_data = True <EOL> return GenericRelatedObjectManager </s>
