<s> '''<STR_LIT>''' <EOL> from weakref import WeakSet <EOL> from sqlalchemy . schema import Column <EOL> from sqlalchemy . orm import object_session <EOL> from sqlalchemy . orm . util import identity_key <EOL> from sqlalchemy . orm . attributes import manager_of_class , QueryableAttribute <EOL> from sqlalchemy . orm . properties import ColumnProperty , RelationshipProperty <EOL> from sqlalchemy . orm . collections import collection_adapter <EOL> from sqlalchemy . orm . attributes import instance_state , instance_dict <EOL> from sqlalchemy . orm . interfaces import MANYTOMANY , MANYTOONE , ONETOMANY <EOL> _included = WeakSet ( ) <EOL> _excluded = WeakSet ( ) <EOL> def include ( prop ) : <EOL> '''<STR_LIT>''' <EOL> if isinstance ( prop , QueryableAttribute ) : <EOL> prop = prop . property <EOL> assert isinstance ( prop , ( Column , ColumnProperty , RelationshipProperty ) ) <EOL> _included . add ( prop ) <EOL> def exclude ( prop ) : <EOL> '''<STR_LIT>''' <EOL> if isinstance ( prop , QueryableAttribute ) : <EOL> prop = prop . property <EOL> assert isinstance ( prop , ( Column , ColumnProperty , RelationshipProperty ) ) <EOL> _excluded . add ( prop ) <EOL> if isinstance ( prop , RelationshipProperty ) : <EOL> for local in prop . local_columns : <EOL> _excluded . add ( local ) <EOL> def reflect ( source , model , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> if source is None : <EOL> return None <EOL> if cache and source in cache : <EOL> return cache [ source ] <EOL> db = object_session ( source ) <EOL> ident = identity_key ( instance = source ) [ <NUM_LIT:1> ] <EOL> assert ident is not None <EOL> return db . query ( model ) . get ( ident ) <EOL> class _PrimaryKeyIsNull ( BaseException ) : <EOL> '''<STR_LIT>''' <EOL> def replicate_relation ( source , target , attr , target_attr , cache = None ) : <EOL> if attr . property . cascade . delete_orphan : <EOL> process_scalar = replicate_no_merge <EOL> process_list = replicate_filter <EOL> else : <EOL> process_scalar = reflect <EOL> process_list = reflect_filter <EOL> value = getattr ( source , attr . key ) <EOL> target_attr_model = target_attr . property . mapper . class_ <EOL> if attr . property . uselist : <EOL> adapter = collection_adapter ( value ) <EOL> if adapter : <EOL> value = adapter . adapt_like_to_iterable ( value ) <EOL> reflection = process_list ( value , target_attr_model , cache = cache ) <EOL> impl = instance_state ( target ) . get_impl ( attr . key ) <EOL> impl . _set_iterable ( instance_state ( target ) , <EOL> instance_dict ( target ) , <EOL> reflection ) <EOL> else : <EOL> reflection = process_scalar ( value , target_attr_model , cache = cache ) <EOL> setattr ( target , attr . key , reflection ) <EOL> if ( reflection is None and <EOL> attr . property . direction is MANYTOONE and <EOL> any ( col . primary_key and not col . nullable <EOL> for col in attr . property . local_columns ) ) : <EOL> raise _PrimaryKeyIsNull ( ) <EOL> def is_relation_replicatable ( attr ) : <EOL> if attr . property in _included : <EOL> return True <EOL> elif attr . property in _excluded : <EOL> return False <EOL> elif attr . property . viewonly : <EOL> return False <EOL> elif attr . property . cascade . delete_orphan : <EOL> return True <EOL> elif attr . property . direction is MANYTOMANY : <EOL> assert attr . property . lazy in ( True , False , '<STR_LIT>' ) <EOL> return attr . property . lazy != '<STR_LIT>' <EOL> elif attr . property . direction is MANYTOONE : <EOL> return True <EOL> else : <EOL> assert attr . property . direction is ONETOMANY <EOL> return False <EOL> def _column_property_in_registry ( prop , registry ) : <EOL> if prop in registry : <EOL> return True <EOL> elif len ( prop . columns ) == <NUM_LIT:1> : <EOL> return prop . columns [ <NUM_LIT:0> ] in registry <EOL> else : <EOL> return False <EOL> def replicate_attributes ( source , target , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> target_manager = manager_of_class ( type ( target ) ) <EOL> column_attrs = set ( ) <EOL> relationship_attrs = set ( ) <EOL> relationship_columns = set ( ) <EOL> for attr in manager_of_class ( type ( source ) ) . attributes : <EOL> if attr . key not in target_manager : <EOL> continue <EOL> target_attr = target_manager [ attr . key ] <EOL> if isinstance ( attr . property , ColumnProperty ) : <EOL> assert isinstance ( target_attr . property , ColumnProperty ) <EOL> column_attrs . add ( attr ) <EOL> elif isinstance ( attr . property , RelationshipProperty ) : <EOL> assert isinstance ( target_attr . property , RelationshipProperty ) <EOL> relationship_attrs . add ( attr ) <EOL> if attr . property . direction is MANYTOONE : <EOL> relationship_columns . update ( attr . property . local_columns ) <EOL> for attr in column_attrs : <EOL> if _column_property_in_registry ( attr . property , _excluded ) : <EOL> continue <EOL> elif ( not _column_property_in_registry ( attr . property , _included ) and <EOL> all ( column in relationship_columns <EOL> for column in attr . property . columns ) ) : <EOL> continue <EOL> setattr ( target , attr . key , getattr ( source , attr . key ) ) <EOL> for attr in relationship_attrs : <EOL> target_attr_model = target_manager [ attr . key ] . property . argument <EOL> if not is_relation_replicatable ( attr ) : <EOL> continue <EOL> replicate_relation ( source , target , attr , target_manager [ attr . key ] , <EOL> cache = cache ) <EOL> def replicate_no_merge ( source , model , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> if source is None : <EOL> return None <EOL> if cache is None : <EOL> cache = { } <EOL> elif source in cache : <EOL> return cache [ source ] <EOL> db = object_session ( source ) <EOL> cls , ident = identity_key ( instance = source ) <EOL> target = db . query ( model ) . get ( ident ) <EOL> if target is None : <EOL> target = model ( ) <EOL> cache [ source ] = target <EOL> try : <EOL> replicate_attributes ( source , target , cache = cache ) <EOL> except _PrimaryKeyIsNull : <EOL> return None <EOL> else : <EOL> return target <EOL> def replicate ( source , model , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> target = replicate_no_merge ( source , model , cache = cache ) <EOL> if target is not None : <EOL> db = object_session ( source ) <EOL> target = db . merge ( target ) <EOL> return target <EOL> def replicate_filter ( sources , model , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> targets = [ replicate_no_merge ( source , model , cache = cache ) <EOL> for source in sources ] <EOL> return [ target for target in targets if target is not None ] <EOL> def reflect_filter ( sources , model , cache = None ) : <EOL> '''<STR_LIT>''' <EOL> targets = [ reflect ( source , model , cache = cache ) for source in sources ] <EOL> return [ target for target in targets if target is not None ] </s>
