<s> """<STR_LIT>""" <EOL> import datetime as dt <EOL> import codecs <EOL> from . type_api import TypeEngine , TypeDecorator , to_instance <EOL> from . elements import quoted_name , type_coerce , _defer_name <EOL> from . default_comparator import _DefaultColumnComparator <EOL> from . . import exc , util , processors <EOL> from . base import _bind_or_error , SchemaEventTarget <EOL> from . import operators <EOL> from . . import event <EOL> from . . util import pickle <EOL> import decimal <EOL> if util . jython : <EOL> import array <EOL> class _DateAffinity ( object ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def _expression_adaptations ( self ) : <EOL> raise NotImplementedError ( ) <EOL> class Comparator ( TypeEngine . Comparator ) : <EOL> _blank_dict = util . immutabledict ( ) <EOL> def _adapt_expression ( self , op , other_comparator ) : <EOL> othertype = other_comparator . type . _type_affinity <EOL> return ( <EOL> op , to_instance ( <EOL> self . type . _expression_adaptations . <EOL> get ( op , self . _blank_dict ) . <EOL> get ( othertype , NULLTYPE ) ) <EOL> ) <EOL> comparator_factory = Comparator <EOL> class Concatenable ( object ) : <EOL> """<STR_LIT>""" <EOL> class Comparator ( TypeEngine . Comparator ) : <EOL> def _adapt_expression ( self , op , other_comparator ) : <EOL> if ( op is operators . add and <EOL> isinstance ( <EOL> other_comparator , <EOL> ( Concatenable . Comparator , NullType . Comparator ) <EOL> ) ) : <EOL> return operators . concat_op , self . expr . type <EOL> else : <EOL> return op , self . expr . type <EOL> comparator_factory = Comparator <EOL> class String ( Concatenable , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:string>' <EOL> def __init__ ( self , length = None , collation = None , <EOL> convert_unicode = False , <EOL> unicode_error = None , <EOL> _warn_on_bytestring = False <EOL> ) : <EOL> """<STR_LIT>""" <EOL> if unicode_error is not None and convert_unicode != '<STR_LIT>' : <EOL> raise exc . ArgumentError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . length = length <EOL> self . collation = collation <EOL> self . convert_unicode = convert_unicode <EOL> self . unicode_error = unicode_error <EOL> self . _warn_on_bytestring = _warn_on_bytestring <EOL> def literal_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> value = value . replace ( "<STR_LIT:'>" , "<STR_LIT>" ) <EOL> return "<STR_LIT>" % value <EOL> return process <EOL> def bind_processor ( self , dialect ) : <EOL> if self . convert_unicode or dialect . convert_unicode : <EOL> if dialect . supports_unicode_binds and self . convert_unicode != '<STR_LIT>' : <EOL> if self . _warn_on_bytestring : <EOL> def process ( value ) : <EOL> if isinstance ( value , util . binary_type ) : <EOL> util . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return value <EOL> return process <EOL> else : <EOL> return None <EOL> else : <EOL> encoder = codecs . getencoder ( dialect . encoding ) <EOL> warn_on_bytestring = self . _warn_on_bytestring <EOL> def process ( value ) : <EOL> if isinstance ( value , util . text_type ) : <EOL> return encoder ( value , self . unicode_error ) [ <NUM_LIT:0> ] <EOL> elif warn_on_bytestring and value is not None : <EOL> util . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return value <EOL> return process <EOL> else : <EOL> return None <EOL> def result_processor ( self , dialect , coltype ) : <EOL> wants_unicode = self . convert_unicode or dialect . convert_unicode <EOL> needs_convert = wants_unicode and ( dialect . returns_unicode_strings is not True or <EOL> self . convert_unicode in ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> needs_isinstance = ( <EOL> needs_convert and <EOL> dialect . returns_unicode_strings and <EOL> self . convert_unicode != '<STR_LIT>' <EOL> ) <EOL> if needs_convert : <EOL> to_unicode = processors . to_unicode_processor_factory ( <EOL> dialect . encoding , self . unicode_error ) <EOL> if needs_isinstance : <EOL> return processors . to_conditional_unicode_processor_factory ( <EOL> dialect . encoding , self . unicode_error ) <EOL> else : <EOL> return processors . to_unicode_processor_factory ( <EOL> dialect . encoding , self . unicode_error ) <EOL> else : <EOL> return None <EOL> @ property <EOL> def python_type ( self ) : <EOL> if self . convert_unicode : <EOL> return util . text_type <EOL> else : <EOL> return str <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . STRING <EOL> class Text ( String ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:text>' <EOL> class Unicode ( String ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> super ( Unicode , self ) . __init__ ( length = length , ** kwargs ) <EOL> class UnicodeText ( Text ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> super ( UnicodeText , self ) . __init__ ( length = length , ** kwargs ) <EOL> class Integer ( _DateAffinity , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . NUMBER <EOL> @ property <EOL> def python_type ( self ) : <EOL> return int <EOL> def literal_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> return str ( value ) <EOL> return process <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . add : { <EOL> Date : Date , <EOL> Integer : self . __class__ , <EOL> Numeric : Numeric , <EOL> } , <EOL> operators . mul : { <EOL> Interval : Interval , <EOL> Integer : self . __class__ , <EOL> Numeric : Numeric , <EOL> } , <EOL> operators . div : { <EOL> Integer : self . __class__ , <EOL> Numeric : Numeric , <EOL> } , <EOL> operators . truediv : { <EOL> Integer : self . __class__ , <EOL> Numeric : Numeric , <EOL> } , <EOL> operators . sub : { <EOL> Integer : self . __class__ , <EOL> Numeric : Numeric , <EOL> } , <EOL> } <EOL> class SmallInteger ( Integer ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class BigInteger ( Integer ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class Numeric ( _DateAffinity , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> _default_decimal_return_scale = <NUM_LIT:10> <EOL> def __init__ ( self , precision = None , scale = None , <EOL> decimal_return_scale = None , asdecimal = True ) : <EOL> """<STR_LIT>""" <EOL> self . precision = precision <EOL> self . scale = scale <EOL> self . decimal_return_scale = decimal_return_scale <EOL> self . asdecimal = asdecimal <EOL> @ property <EOL> def _effective_decimal_return_scale ( self ) : <EOL> if self . decimal_return_scale is not None : <EOL> return self . decimal_return_scale <EOL> elif getattr ( self , "<STR_LIT>" , None ) is not None : <EOL> return self . scale <EOL> else : <EOL> return self . _default_decimal_return_scale <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . NUMBER <EOL> def literal_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> return str ( value ) <EOL> return process <EOL> @ property <EOL> def python_type ( self ) : <EOL> if self . asdecimal : <EOL> return decimal . Decimal <EOL> else : <EOL> return float <EOL> def bind_processor ( self , dialect ) : <EOL> if dialect . supports_native_decimal : <EOL> return None <EOL> else : <EOL> return processors . to_float <EOL> def result_processor ( self , dialect , coltype ) : <EOL> if self . asdecimal : <EOL> if dialect . supports_native_decimal : <EOL> return None <EOL> else : <EOL> util . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( dialect . name , dialect . driver ) ) <EOL> return processors . to_decimal_processor_factory ( <EOL> decimal . Decimal , <EOL> self . scale if self . scale is not None <EOL> else self . _default_decimal_return_scale ) <EOL> else : <EOL> if dialect . supports_native_decimal : <EOL> return processors . to_float <EOL> else : <EOL> return None <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . mul : { <EOL> Interval : Interval , <EOL> Numeric : self . __class__ , <EOL> Integer : self . __class__ , <EOL> } , <EOL> operators . div : { <EOL> Numeric : self . __class__ , <EOL> Integer : self . __class__ , <EOL> } , <EOL> operators . truediv : { <EOL> Numeric : self . __class__ , <EOL> Integer : self . __class__ , <EOL> } , <EOL> operators . add : { <EOL> Numeric : self . __class__ , <EOL> Integer : self . __class__ , <EOL> } , <EOL> operators . sub : { <EOL> Numeric : self . __class__ , <EOL> Integer : self . __class__ , <EOL> } <EOL> } <EOL> class Float ( Numeric ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:float>' <EOL> scale = None <EOL> def __init__ ( self , precision = None , asdecimal = False , <EOL> decimal_return_scale = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . precision = precision <EOL> self . asdecimal = asdecimal <EOL> self . decimal_return_scale = decimal_return_scale <EOL> if kwargs : <EOL> util . warn_deprecated ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def result_processor ( self , dialect , coltype ) : <EOL> if self . asdecimal : <EOL> return processors . to_decimal_processor_factory ( <EOL> decimal . Decimal , <EOL> self . _effective_decimal_return_scale ) <EOL> else : <EOL> return None <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . mul : { <EOL> Interval : Interval , <EOL> Numeric : self . __class__ , <EOL> } , <EOL> operators . div : { <EOL> Numeric : self . __class__ , <EOL> } , <EOL> operators . truediv : { <EOL> Numeric : self . __class__ , <EOL> } , <EOL> operators . add : { <EOL> Numeric : self . __class__ , <EOL> } , <EOL> operators . sub : { <EOL> Numeric : self . __class__ , <EOL> } <EOL> } <EOL> class DateTime ( _DateAffinity , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , timezone = False ) : <EOL> """<STR_LIT>""" <EOL> self . timezone = timezone <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . DATETIME <EOL> @ property <EOL> def python_type ( self ) : <EOL> return dt . datetime <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . add : { <EOL> Interval : self . __class__ , <EOL> } , <EOL> operators . sub : { <EOL> Interval : self . __class__ , <EOL> DateTime : Interval , <EOL> } , <EOL> } <EOL> class Date ( _DateAffinity , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:date>' <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . DATETIME <EOL> @ property <EOL> def python_type ( self ) : <EOL> return dt . date <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . add : { <EOL> Integer : self . __class__ , <EOL> Interval : DateTime , <EOL> Time : DateTime , <EOL> } , <EOL> operators . sub : { <EOL> Integer : self . __class__ , <EOL> Date : Integer , <EOL> Interval : DateTime , <EOL> DateTime : Interval , <EOL> } , <EOL> } <EOL> class Time ( _DateAffinity , TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:time>' <EOL> def __init__ ( self , timezone = False ) : <EOL> self . timezone = timezone <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . DATETIME <EOL> @ property <EOL> def python_type ( self ) : <EOL> return dt . time <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . add : { <EOL> Date : DateTime , <EOL> Interval : self . __class__ <EOL> } , <EOL> operators . sub : { <EOL> Time : Interval , <EOL> Interval : self . __class__ , <EOL> } , <EOL> } <EOL> class _Binary ( TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , length = None ) : <EOL> self . length = length <EOL> def literal_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> value = value . decode ( dialect . encoding ) . replace ( "<STR_LIT:'>" , "<STR_LIT>" ) <EOL> return "<STR_LIT>" % value <EOL> return process <EOL> @ property <EOL> def python_type ( self ) : <EOL> return util . binary_type <EOL> def bind_processor ( self , dialect ) : <EOL> if dialect . dbapi is None : <EOL> return None <EOL> DBAPIBinary = dialect . dbapi . Binary <EOL> def process ( value ) : <EOL> if value is not None : <EOL> return DBAPIBinary ( value ) <EOL> else : <EOL> return None <EOL> return process <EOL> if util . py2k : <EOL> def result_processor ( self , dialect , coltype ) : <EOL> if util . jython : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> if isinstance ( value , array . array ) : <EOL> return value . tostring ( ) <EOL> return str ( value ) <EOL> else : <EOL> return None <EOL> else : <EOL> process = processors . to_str <EOL> return process <EOL> else : <EOL> def result_processor ( self , dialect , coltype ) : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = bytes ( value ) <EOL> return value <EOL> return process <EOL> def coerce_compared_value ( self , op , value ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( value , util . string_types ) : <EOL> return self <EOL> else : <EOL> return super ( _Binary , self ) . coerce_compared_value ( op , value ) <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . BINARY <EOL> class LargeBinary ( _Binary ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None ) : <EOL> """<STR_LIT>""" <EOL> _Binary . __init__ ( self , length = length ) <EOL> class Binary ( LargeBinary ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * arg , ** kw ) : <EOL> util . warn_deprecated ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> LargeBinary . __init__ ( self , * arg , ** kw ) <EOL> class SchemaType ( SchemaEventTarget ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name = None , schema = None , metadata = None , <EOL> inherit_schema = False , quote = None ) : <EOL> if name is not None : <EOL> self . name = quoted_name ( name , quote ) <EOL> else : <EOL> self . name = None <EOL> self . schema = schema <EOL> self . metadata = metadata <EOL> self . inherit_schema = inherit_schema <EOL> if self . metadata : <EOL> event . listen ( <EOL> self . metadata , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( self . _on_metadata_create ) <EOL> ) <EOL> event . listen ( <EOL> self . metadata , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( self . _on_metadata_drop ) <EOL> ) <EOL> def _set_parent ( self , column ) : <EOL> column . _on_table_attach ( util . portable_instancemethod ( self . _set_table ) ) <EOL> def _set_table ( self , column , table ) : <EOL> if self . inherit_schema : <EOL> self . schema = table . schema <EOL> event . listen ( <EOL> table , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( <EOL> self . _on_table_create ) <EOL> ) <EOL> event . listen ( <EOL> table , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( self . _on_table_drop ) <EOL> ) <EOL> if self . metadata is None : <EOL> event . listen ( <EOL> table . metadata , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( self . _on_metadata_create ) <EOL> ) <EOL> event . listen ( <EOL> table . metadata , <EOL> "<STR_LIT>" , <EOL> util . portable_instancemethod ( self . _on_metadata_drop ) <EOL> ) <EOL> def copy ( self , ** kw ) : <EOL> return self . adapt ( self . __class__ ) <EOL> def adapt ( self , impltype , ** kw ) : <EOL> schema = kw . pop ( '<STR_LIT>' , self . schema ) <EOL> metadata = kw . pop ( '<STR_LIT>' , None ) <EOL> return impltype ( name = self . name , <EOL> schema = schema , <EOL> metadata = metadata , <EOL> inherit_schema = self . inherit_schema , <EOL> ** kw ) <EOL> @ property <EOL> def bind ( self ) : <EOL> return self . metadata and self . metadata . bind or None <EOL> def create ( self , bind = None , checkfirst = False ) : <EOL> """<STR_LIT>""" <EOL> if bind is None : <EOL> bind = _bind_or_error ( self ) <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . create ( bind = bind , checkfirst = checkfirst ) <EOL> def drop ( self , bind = None , checkfirst = False ) : <EOL> """<STR_LIT>""" <EOL> if bind is None : <EOL> bind = _bind_or_error ( self ) <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . drop ( bind = bind , checkfirst = checkfirst ) <EOL> def _on_table_create ( self , target , bind , ** kw ) : <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . _on_table_create ( target , bind , ** kw ) <EOL> def _on_table_drop ( self , target , bind , ** kw ) : <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . _on_table_drop ( target , bind , ** kw ) <EOL> def _on_metadata_create ( self , target , bind , ** kw ) : <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . _on_metadata_create ( target , bind , ** kw ) <EOL> def _on_metadata_drop ( self , target , bind , ** kw ) : <EOL> t = self . dialect_impl ( bind . dialect ) <EOL> if t . __class__ is not self . __class__ and isinstance ( t , SchemaType ) : <EOL> t . _on_metadata_drop ( target , bind , ** kw ) <EOL> class Enum ( String , SchemaType ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , * enums , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . enums = enums <EOL> self . native_enum = kw . pop ( '<STR_LIT>' , True ) <EOL> convert_unicode = kw . pop ( '<STR_LIT>' , None ) <EOL> if convert_unicode is None : <EOL> for e in enums : <EOL> if isinstance ( e , util . text_type ) : <EOL> convert_unicode = True <EOL> break <EOL> else : <EOL> convert_unicode = False <EOL> if self . enums : <EOL> length = max ( len ( x ) for x in self . enums ) <EOL> else : <EOL> length = <NUM_LIT:0> <EOL> String . __init__ ( self , <EOL> length = length , <EOL> convert_unicode = convert_unicode , <EOL> ) <EOL> SchemaType . __init__ ( self , ** kw ) <EOL> def __repr__ ( self ) : <EOL> return util . generic_repr ( self , <EOL> to_inspect = [ Enum , SchemaType ] , <EOL> ) <EOL> def _should_create_constraint ( self , compiler ) : <EOL> return not self . native_enum or not compiler . dialect . supports_native_enum <EOL> @ util . dependencies ( "<STR_LIT>" ) <EOL> def _set_table ( self , schema , column , table ) : <EOL> if self . native_enum : <EOL> SchemaType . _set_table ( self , column , table ) <EOL> e = schema . CheckConstraint ( <EOL> type_coerce ( column , self ) . in_ ( self . enums ) , <EOL> name = _defer_name ( self . name ) , <EOL> _create_rule = util . portable_instancemethod ( <EOL> self . _should_create_constraint ) <EOL> ) <EOL> assert e . table is table <EOL> def adapt ( self , impltype , ** kw ) : <EOL> schema = kw . pop ( '<STR_LIT>' , self . schema ) <EOL> metadata = kw . pop ( '<STR_LIT>' , None ) <EOL> if issubclass ( impltype , Enum ) : <EOL> return impltype ( name = self . name , <EOL> schema = schema , <EOL> metadata = metadata , <EOL> convert_unicode = self . convert_unicode , <EOL> native_enum = self . native_enum , <EOL> inherit_schema = self . inherit_schema , <EOL> * self . enums , <EOL> ** kw ) <EOL> else : <EOL> return super ( Enum , self ) . adapt ( impltype , ** kw ) <EOL> class PickleType ( TypeDecorator ) : <EOL> """<STR_LIT>""" <EOL> impl = LargeBinary <EOL> def __init__ ( self , protocol = pickle . HIGHEST_PROTOCOL , <EOL> pickler = None , comparator = None ) : <EOL> """<STR_LIT>""" <EOL> self . protocol = protocol <EOL> self . pickler = pickler or pickle <EOL> self . comparator = comparator <EOL> super ( PickleType , self ) . __init__ ( ) <EOL> def __reduce__ ( self ) : <EOL> return PickleType , ( self . protocol , <EOL> None , <EOL> self . comparator ) <EOL> def bind_processor ( self , dialect ) : <EOL> impl_processor = self . impl . bind_processor ( dialect ) <EOL> dumps = self . pickler . dumps <EOL> protocol = self . protocol <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = dumps ( value , protocol ) <EOL> return impl_processor ( value ) <EOL> else : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = dumps ( value , protocol ) <EOL> return value <EOL> return process <EOL> def result_processor ( self , dialect , coltype ) : <EOL> impl_processor = self . impl . result_processor ( dialect , coltype ) <EOL> loads = self . pickler . loads <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> value = impl_processor ( value ) <EOL> if value is None : <EOL> return None <EOL> return loads ( value ) <EOL> else : <EOL> def process ( value ) : <EOL> if value is None : <EOL> return None <EOL> return loads ( value ) <EOL> return process <EOL> def compare_values ( self , x , y ) : <EOL> if self . comparator : <EOL> return self . comparator ( x , y ) <EOL> else : <EOL> return x == y <EOL> class Boolean ( TypeEngine , SchemaType ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , create_constraint = True , name = None ) : <EOL> """<STR_LIT>""" <EOL> self . create_constraint = create_constraint <EOL> self . name = name <EOL> def _should_create_constraint ( self , compiler ) : <EOL> return not compiler . dialect . supports_native_boolean <EOL> @ util . dependencies ( "<STR_LIT>" ) <EOL> def _set_table ( self , schema , column , table ) : <EOL> if not self . create_constraint : <EOL> return <EOL> e = schema . CheckConstraint ( <EOL> type_coerce ( column , self ) . in_ ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) , <EOL> name = _defer_name ( self . name ) , <EOL> _create_rule = util . portable_instancemethod ( <EOL> self . _should_create_constraint ) <EOL> ) <EOL> assert e . table is table <EOL> @ property <EOL> def python_type ( self ) : <EOL> return bool <EOL> def literal_processor ( self , dialect ) : <EOL> if dialect . supports_native_boolean : <EOL> def process ( value ) : <EOL> return "<STR_LIT:true>" if value else "<STR_LIT:false>" <EOL> else : <EOL> def process ( value ) : <EOL> return str ( <NUM_LIT:1> if value else <NUM_LIT:0> ) <EOL> return process <EOL> def bind_processor ( self , dialect ) : <EOL> if dialect . supports_native_boolean : <EOL> return None <EOL> else : <EOL> return processors . boolean_to_int <EOL> def result_processor ( self , dialect , coltype ) : <EOL> if dialect . supports_native_boolean : <EOL> return None <EOL> else : <EOL> return processors . int_to_boolean <EOL> class Interval ( _DateAffinity , TypeDecorator ) : <EOL> """<STR_LIT>""" <EOL> impl = DateTime <EOL> epoch = dt . datetime . utcfromtimestamp ( <NUM_LIT:0> ) <EOL> def __init__ ( self , native = True , <EOL> second_precision = None , <EOL> day_precision = None ) : <EOL> """<STR_LIT>""" <EOL> super ( Interval , self ) . __init__ ( ) <EOL> self . native = native <EOL> self . second_precision = second_precision <EOL> self . day_precision = day_precision <EOL> def adapt ( self , cls , ** kw ) : <EOL> if self . native and hasattr ( cls , '<STR_LIT>' ) : <EOL> return cls . _adapt_from_generic_interval ( self , ** kw ) <EOL> else : <EOL> return self . __class__ ( <EOL> native = self . native , <EOL> second_precision = self . second_precision , <EOL> day_precision = self . day_precision , <EOL> ** kw ) <EOL> @ property <EOL> def python_type ( self ) : <EOL> return dt . timedelta <EOL> def bind_processor ( self , dialect ) : <EOL> impl_processor = self . impl . bind_processor ( dialect ) <EOL> epoch = self . epoch <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = epoch + value <EOL> return impl_processor ( value ) <EOL> else : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = epoch + value <EOL> return value <EOL> return process <EOL> def result_processor ( self , dialect , coltype ) : <EOL> impl_processor = self . impl . result_processor ( dialect , coltype ) <EOL> epoch = self . epoch <EOL> if impl_processor : <EOL> def process ( value ) : <EOL> value = impl_processor ( value ) <EOL> if value is None : <EOL> return None <EOL> return value - epoch <EOL> else : <EOL> def process ( value ) : <EOL> if value is None : <EOL> return None <EOL> return value - epoch <EOL> return process <EOL> @ util . memoized_property <EOL> def _expression_adaptations ( self ) : <EOL> return { <EOL> operators . add : { <EOL> Date : DateTime , <EOL> Interval : self . __class__ , <EOL> DateTime : DateTime , <EOL> Time : Time , <EOL> } , <EOL> operators . sub : { <EOL> Interval : self . __class__ <EOL> } , <EOL> operators . mul : { <EOL> Numeric : self . __class__ <EOL> } , <EOL> operators . truediv : { <EOL> Numeric : self . __class__ <EOL> } , <EOL> operators . div : { <EOL> Numeric : self . __class__ <EOL> } <EOL> } <EOL> @ property <EOL> def _type_affinity ( self ) : <EOL> return Interval <EOL> def coerce_compared_value ( self , op , value ) : <EOL> """<STR_LIT>""" <EOL> return self . impl . coerce_compared_value ( op , value ) <EOL> class REAL ( Float ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class FLOAT ( Float ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class NUMERIC ( Numeric ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class DECIMAL ( Numeric ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class INTEGER ( Integer ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> INT = INTEGER <EOL> class SMALLINT ( SmallInteger ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class BIGINT ( BigInteger ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class TIMESTAMP ( DateTime ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def get_dbapi_type ( self , dbapi ) : <EOL> return dbapi . TIMESTAMP <EOL> class DATETIME ( DateTime ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class DATE ( Date ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class TIME ( Time ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class TEXT ( Text ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class CLOB ( Text ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class VARCHAR ( String ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class NVARCHAR ( Unicode ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class CHAR ( String ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class NCHAR ( Unicode ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class BLOB ( LargeBinary ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class BINARY ( _Binary ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class VARBINARY ( _Binary ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class BOOLEAN ( Boolean ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class NullType ( TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:null>' <EOL> _isnull = True <EOL> def literal_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> return "<STR_LIT>" <EOL> return process <EOL> class Comparator ( TypeEngine . Comparator ) : <EOL> def _adapt_expression ( self , op , other_comparator ) : <EOL> if isinstance ( other_comparator , NullType . Comparator ) or not operators . is_commutative ( op ) : <EOL> return op , self . expr . type <EOL> else : <EOL> return other_comparator . _adapt_expression ( op , self ) <EOL> comparator_factory = Comparator <EOL> NULLTYPE = NullType ( ) <EOL> BOOLEANTYPE = Boolean ( ) <EOL> STRINGTYPE = String ( ) <EOL> INTEGERTYPE = Integer ( ) <EOL> _type_map = { <EOL> int : Integer ( ) , <EOL> float : Numeric ( ) , <EOL> bool : BOOLEANTYPE , <EOL> decimal . Decimal : Numeric ( ) , <EOL> dt . date : Date ( ) , <EOL> dt . datetime : DateTime ( ) , <EOL> dt . time : Time ( ) , <EOL> dt . timedelta : Interval ( ) , <EOL> util . NoneType : NULLTYPE <EOL> } <EOL> if util . py3k : <EOL> _type_map [ bytes ] = LargeBinary ( ) <EOL> _type_map [ str ] = Unicode ( ) <EOL> else : <EOL> _type_map [ unicode ] = Unicode ( ) <EOL> _type_map [ str ] = String ( ) <EOL> from . import type_api <EOL> type_api . BOOLEANTYPE = BOOLEANTYPE <EOL> type_api . STRINGTYPE = STRINGTYPE <EOL> type_api . INTEGERTYPE = INTEGERTYPE <EOL> type_api . NULLTYPE = NULLTYPE <EOL> type_api . _type_map = _type_map <EOL> class Comparator ( _DefaultColumnComparator ) : <EOL> BOOLEANTYPE = BOOLEANTYPE <EOL> TypeEngine . Comparator . __bases__ = ( <EOL> Comparator , ) + TypeEngine . Comparator . __bases__ </s>
