<s> """<STR_LIT>""" <EOL> from . . import exc , util <EOL> from . import operators <EOL> from . visitors import Visitable <EOL> BOOLEANTYPE = None <EOL> INTEGERTYPE = None <EOL> NULLTYPE = None <EOL> STRINGTYPE = None <EOL> class TypeEngine ( Visitable ) : <EOL> """<STR_LIT>""" <EOL> _sqla_type = True <EOL> _isnull = False <EOL> class Comparator ( operators . ColumnOperators ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , expr ) : <EOL> self . expr = expr <EOL> def __reduce__ ( self ) : <EOL> return _reconstitute_comparator , ( self . expr , ) <EOL> hashable = True <EOL> """<STR_LIT>""" <EOL> comparator_factory = Comparator <EOL> """<STR_LIT>""" <EOL> def copy_value ( self , value ) : <EOL> return value <EOL> def literal_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def bind_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def result_processor ( self , dialect , coltype ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def column_expression ( self , colexpr ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> @ util . memoized_property <EOL> def _has_column_expression ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ . column_expression . __code__ is not TypeEngine . column_expression . __code__ <EOL> def bind_expression ( self , bindvalue ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> @ util . memoized_property <EOL> def _has_bind_expression ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ . bind_expression . __code__ is not TypeEngine . bind_expression . __code__ <EOL> def compare_values ( self , x , y ) : <EOL> """<STR_LIT>""" <EOL> return x == y <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> @ property <EOL> def python_type ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def with_variant ( self , type_ , dialect_name ) : <EOL> """<STR_LIT>""" <EOL> return Variant ( self , { dialect_name : to_instance ( type_ ) } ) <EOL> @ util . memoized_property <EOL> def _type_affinity ( self ) : <EOL> """<STR_LIT>""" <EOL> typ = None <EOL> for t in self . __class__ . __mro__ : <EOL> if t in ( TypeEngine , UserDefinedType ) : <EOL> return typ <EOL> elif issubclass ( t , ( TypeEngine , UserDefinedType ) ) : <EOL> typ = t <EOL> else : <EOL> return self . __class__ <EOL> def dialect_impl ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return dialect . _type_memos [ self ] [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> return self . _dialect_info ( dialect ) [ '<STR_LIT>' ] <EOL> def _cached_literal_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return dialect . _type_memos [ self ] [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> d = self . _dialect_info ( dialect ) <EOL> d [ '<STR_LIT>' ] = lp = d [ '<STR_LIT>' ] . literal_processor ( dialect ) <EOL> return lp <EOL> def _cached_bind_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return dialect . _type_memos [ self ] [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> d = self . _dialect_info ( dialect ) <EOL> d [ '<STR_LIT>' ] = bp = d [ '<STR_LIT>' ] . bind_processor ( dialect ) <EOL> return bp <EOL> def _cached_result_processor ( self , dialect , coltype ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return dialect . _type_memos [ self ] [ coltype ] <EOL> except KeyError : <EOL> d = self . _dialect_info ( dialect ) <EOL> d [ coltype ] = rp = d [ '<STR_LIT>' ] . result_processor ( dialect , coltype ) <EOL> return rp <EOL> def _dialect_info ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> if self in dialect . _type_memos : <EOL> return dialect . _type_memos [ self ] <EOL> else : <EOL> impl = self . _gen_dialect_impl ( dialect ) <EOL> if impl is self : <EOL> impl = self . adapt ( type ( self ) ) <EOL> assert impl is not self <EOL> dialect . _type_memos [ self ] = d = { '<STR_LIT>' : impl } <EOL> return d <EOL> def _gen_dialect_impl ( self , dialect ) : <EOL> return dialect . type_descriptor ( self ) <EOL> def adapt ( self , cls , ** kw ) : <EOL> """<STR_LIT>""" <EOL> return util . constructor_copy ( self , cls , ** kw ) <EOL> def coerce_compared_value ( self , op , value ) : <EOL> """<STR_LIT>""" <EOL> _coerced_type = _type_map . get ( type ( value ) , NULLTYPE ) <EOL> if _coerced_type is NULLTYPE or _coerced_type . _type_affinity is self . _type_affinity : <EOL> return self <EOL> else : <EOL> return _coerced_type <EOL> def _compare_type_affinity ( self , other ) : <EOL> return self . _type_affinity is other . _type_affinity <EOL> def compile ( self , dialect = None ) : <EOL> """<STR_LIT>""" <EOL> if not dialect : <EOL> dialect = self . _default_dialect ( ) <EOL> return dialect . type_compiler . process ( self ) <EOL> @ util . dependencies ( "<STR_LIT>" ) <EOL> def _default_dialect ( self , default ) : <EOL> if self . __class__ . __module__ . startswith ( "<STR_LIT>" ) : <EOL> tokens = self . __class__ . __module__ . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> : <NUM_LIT:3> ] <EOL> mod = "<STR_LIT:.>" . join ( tokens ) <EOL> return getattr ( __import__ ( mod ) . dialects , tokens [ - <NUM_LIT:1> ] ) . dialect ( ) <EOL> else : <EOL> return default . DefaultDialect ( ) <EOL> def __str__ ( self ) : <EOL> if util . py2k : <EOL> return unicode ( self . compile ( ) ) . encode ( '<STR_LIT:ascii>' , '<STR_LIT>' ) <EOL> else : <EOL> return str ( self . compile ( ) ) <EOL> def __repr__ ( self ) : <EOL> return util . generic_repr ( self ) <EOL> class UserDefinedType ( TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = "<STR_LIT>" <EOL> class Comparator ( TypeEngine . Comparator ) : <EOL> def _adapt_expression ( self , op , other_comparator ) : <EOL> if hasattr ( self . type , '<STR_LIT>' ) : <EOL> util . warn_deprecated ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> return self . type . adapt_operator ( op ) , self . type <EOL> else : <EOL> return op , self . type <EOL> comparator_factory = Comparator <EOL> def coerce_compared_value ( self , op , value ) : <EOL> """<STR_LIT>""" <EOL> return self <EOL> class TypeDecorator ( TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = "<STR_LIT>" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self . __class__ , '<STR_LIT>' ) : <EOL> raise AssertionError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . impl = to_instance ( self . __class__ . impl , * args , ** kwargs ) <EOL> coerce_to_is_types = ( util . NoneType , ) <EOL> """<STR_LIT>""" <EOL> class Comparator ( TypeEngine . Comparator ) : <EOL> def operate ( self , op , * other , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = self . expr . type . coerce_to_is_types <EOL> return super ( TypeDecorator . Comparator , self ) . operate ( <EOL> op , * other , ** kwargs ) <EOL> def reverse_operate ( self , op , other , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = self . expr . type . coerce_to_is_types <EOL> return super ( TypeDecorator . Comparator , self ) . reverse_operate ( <EOL> op , other , ** kwargs ) <EOL> @ property <EOL> def comparator_factory ( self ) : <EOL> return type ( "<STR_LIT>" , <EOL> ( TypeDecorator . Comparator , self . impl . comparator_factory ) , <EOL> { } ) <EOL> def _gen_dialect_impl ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> adapted = dialect . type_descriptor ( self ) <EOL> if adapted is not self : <EOL> return adapted <EOL> typedesc = self . load_dialect_impl ( dialect ) . dialect_impl ( dialect ) <EOL> tt = self . copy ( ) <EOL> if not isinstance ( tt , self . __class__ ) : <EOL> raise AssertionError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( self , self . __class__ ) ) <EOL> tt . impl = typedesc <EOL> return tt <EOL> @ property <EOL> def _type_affinity ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . impl . _type_affinity <EOL> def type_engine ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> adapted = dialect . type_descriptor ( self ) <EOL> if not isinstance ( adapted , type ( self ) ) : <EOL> return adapted <EOL> elif isinstance ( self . impl , TypeDecorator ) : <EOL> return self . impl . type_engine ( dialect ) <EOL> else : <EOL> return self . load_dialect_impl ( dialect ) <EOL> def load_dialect_impl ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> return self . impl <EOL> def __getattr__ ( self , key ) : <EOL> """<STR_LIT>""" <EOL> return getattr ( self . impl , key ) <EOL> def process_literal_param ( self , value , dialect ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def process_bind_param ( self , value , dialect ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def process_result_value ( self , value , dialect ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> @ util . memoized_property <EOL> def _has_bind_processor ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ . process_bind_param . __code__ is not TypeDecorator . process_bind_param . __code__ <EOL> @ util . memoized_property <EOL> def _has_literal_processor ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ . process_literal_param . __code__ is not TypeDecorator . process_literal_param . __code__ <EOL> def literal_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> if self . _has_literal_processor : <EOL> process_param = self . process_literal_param <EOL> elif self . _has_bind_processor : <EOL> process_param = self . process_bind_param <EOL> else : <EOL> process_param = None <EOL> if process_param : <EOL> impl_processor = self . impl . literal_processor ( dialect ) <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> return impl_processor ( process_param ( value , dialect ) ) <EOL> else : <EOL> def process ( value ) : <EOL> return process_param ( value , dialect ) <EOL> return process <EOL> else : <EOL> return self . impl . literal_processor ( dialect ) <EOL> def bind_processor ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> if self . _has_bind_processor : <EOL> process_param = self . process_bind_param <EOL> impl_processor = self . impl . bind_processor ( dialect ) <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> return impl_processor ( process_param ( value , dialect ) ) <EOL> else : <EOL> def process ( value ) : <EOL> return process_param ( value , dialect ) <EOL> return process <EOL> else : <EOL> return self . impl . bind_processor ( dialect ) <EOL> @ util . memoized_property <EOL> def _has_result_processor ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ . process_result_value . __code__ is not TypeDecorator . process_result_value . __code__ <EOL> def result_processor ( self , dialect , coltype ) : <EOL> """<STR_LIT>""" <EOL> if self . _has_result_processor : <EOL> process_value = self . process_result_value <EOL> impl_processor = self . impl . result_processor ( dialect , <EOL> coltype ) <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> return process_value ( impl_processor ( value ) , dialect ) <EOL> else : <EOL> def process ( value ) : <EOL> return process_value ( value , dialect ) <EOL> return process <EOL> else : <EOL> return self . impl . result_processor ( dialect , coltype ) <EOL> def coerce_compared_value ( self , op , value ) : <EOL> """<STR_LIT>""" <EOL> return self <EOL> def copy ( self ) : <EOL> """<STR_LIT>""" <EOL> instance = self . __class__ . __new__ ( self . __class__ ) <EOL> instance . __dict__ . update ( self . __dict__ ) <EOL> return instance <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> """<STR_LIT>""" <EOL> return self . impl . get_dbapi_type ( dbapi ) <EOL> def compare_values ( self , x , y ) : <EOL> """<STR_LIT>""" <EOL> return self . impl . compare_values ( x , y ) <EOL> def __repr__ ( self ) : <EOL> return util . generic_repr ( self , to_inspect = self . impl ) <EOL> class Variant ( TypeDecorator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , base , mapping ) : <EOL> """<STR_LIT>""" <EOL> self . impl = base <EOL> self . mapping = mapping <EOL> def load_dialect_impl ( self , dialect ) : <EOL> if dialect . name in self . mapping : <EOL> return self . mapping [ dialect . name ] <EOL> else : <EOL> return self . impl <EOL> def with_variant ( self , type_ , dialect_name ) : <EOL> """<STR_LIT>""" <EOL> if dialect_name in self . mapping : <EOL> raise exc . ArgumentError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % dialect_name ) <EOL> mapping = self . mapping . copy ( ) <EOL> mapping [ dialect_name ] = type_ <EOL> return Variant ( self . impl , mapping ) <EOL> @ property <EOL> def comparator_factory ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . impl . comparator_factory <EOL> def _reconstitute_comparator ( expression ) : <EOL> return expression . comparator <EOL> def to_instance ( typeobj , * arg , ** kw ) : <EOL> if typeobj is None : <EOL> return NULLTYPE <EOL> if util . callable ( typeobj ) : <EOL> return typeobj ( * arg , ** kw ) <EOL> else : <EOL> return typeobj <EOL> def adapt_type ( typeobj , colspecs ) : <EOL> if isinstance ( typeobj , type ) : <EOL> typeobj = typeobj ( ) <EOL> for t in typeobj . __class__ . __mro__ [ <NUM_LIT:0> : - <NUM_LIT:1> ] : <EOL> try : <EOL> impltype = colspecs [ t ] <EOL> break <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> return typeobj <EOL> if ( issubclass ( typeobj . __class__ , impltype ) ) : <EOL> return typeobj <EOL> return typeobj . adapt ( impltype ) </s>
