<s> from . _compat import PY2 , text_type , long_type , JYTHON , IRONPYTHON <EOL> import sys <EOL> import binascii <EOL> import datetime <EOL> from decimal import Decimal <EOL> import re <EOL> import time <EOL> from . constants import FIELD_TYPE , FLAG <EOL> from . charset import charset_by_id , charset_to_encoding <EOL> ESCAPE_REGEX = re . compile ( r"<STR_LIT>" ) <EOL> ESCAPE_MAP = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:\n>' : '<STR_LIT>' , '<STR_LIT:\r>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:">' : '<STR_LIT>' , '<STR_LIT:\\>' : '<STR_LIT>' } <EOL> def escape_item ( val , charset ) : <EOL> if type ( val ) in [ tuple , list , set ] : <EOL> return escape_sequence ( val , charset ) <EOL> if type ( val ) is dict : <EOL> return escape_dict ( val , charset ) <EOL> encoder = encoders [ type ( val ) ] <EOL> val = encoder ( val ) <EOL> if type ( val ) in [ str , int , text_type ] : <EOL> return val <EOL> return val <EOL> def escape_dict ( val , charset ) : <EOL> n = { } <EOL> for k , v in val . items ( ) : <EOL> quoted = escape_item ( v , charset ) <EOL> n [ k ] = quoted <EOL> return n <EOL> def escape_sequence ( val , charset ) : <EOL> n = [ ] <EOL> for item in val : <EOL> quoted = escape_item ( item , charset ) <EOL> n . append ( quoted ) <EOL> return "<STR_LIT:(>" + "<STR_LIT:U+002C>" . join ( n ) + "<STR_LIT:)>" <EOL> def escape_set ( val , charset ) : <EOL> val = map ( lambda x : escape_item ( x , charset ) , val ) <EOL> return '<STR_LIT:U+002C>' . join ( val ) <EOL> def escape_bool ( value ) : <EOL> return str ( int ( value ) ) <EOL> def escape_object ( value ) : <EOL> return str ( value ) <EOL> def escape_int ( value ) : <EOL> return str ( value ) <EOL> def escape_float ( value ) : <EOL> return ( '<STR_LIT>' % value ) <EOL> def escape_string ( value ) : <EOL> return ( "<STR_LIT:%s>" % ( ESCAPE_REGEX . sub ( <EOL> lambda match : ESCAPE_MAP . get ( match . group ( <NUM_LIT:0> ) ) , value ) , ) ) <EOL> def escape_str ( value ) : <EOL> return "<STR_LIT>" % escape_string ( value ) <EOL> def escape_unicode ( value ) : <EOL> return escape_str ( value ) <EOL> def escape_bytes ( value ) : <EOL> return "<STR_LIT>" % binascii . hexlify ( value ) . decode ( sys . getdefaultencoding ( ) ) <EOL> def escape_None ( value ) : <EOL> return '<STR_LIT>' <EOL> def escape_timedelta ( obj ) : <EOL> seconds = int ( obj . seconds ) % <NUM_LIT> <EOL> minutes = int ( obj . seconds // <NUM_LIT> ) % <NUM_LIT> <EOL> hours = int ( obj . seconds // <NUM_LIT> ) % <NUM_LIT> + int ( obj . days ) * <NUM_LIT> <EOL> return escape_str ( '<STR_LIT>' % ( hours , minutes , seconds ) ) <EOL> def escape_time ( obj ) : <EOL> s = "<STR_LIT>" % ( int ( obj . hour ) , int ( obj . minute ) , <EOL> int ( obj . second ) ) <EOL> if obj . microsecond : <EOL> s += "<STR_LIT>" . format ( obj . microsecond ) <EOL> return escape_str ( s ) <EOL> def escape_datetime ( obj ) : <EOL> return escape_str ( obj . isoformat ( '<STR_LIT:U+0020>' ) ) <EOL> def escape_date ( obj ) : <EOL> return escape_str ( obj . isoformat ( ) ) <EOL> def escape_struct_time ( obj ) : <EOL> return escape_datetime ( datetime . datetime ( * obj [ : <NUM_LIT:6> ] ) ) <EOL> def convert_datetime ( obj ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:U+0020>' in obj : <EOL> sep = '<STR_LIT:U+0020>' <EOL> elif '<STR_LIT:T>' in obj : <EOL> sep = '<STR_LIT:T>' <EOL> else : <EOL> return convert_date ( obj ) <EOL> try : <EOL> ymd , hms = obj . split ( sep , <NUM_LIT:1> ) <EOL> usecs = '<STR_LIT:0>' <EOL> if '<STR_LIT:.>' in hms : <EOL> hms , usecs = hms . split ( '<STR_LIT:.>' ) <EOL> usecs = float ( '<STR_LIT>' + usecs ) * <NUM_LIT> <EOL> return datetime . datetime ( * [ int ( x ) for x in ymd . split ( '<STR_LIT:->' ) + hms . split ( '<STR_LIT::>' ) + [ usecs ] ] ) <EOL> except ValueError : <EOL> return convert_date ( obj ) <EOL> def convert_timedelta ( obj ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> microseconds = <NUM_LIT:0> <EOL> if "<STR_LIT:.>" in obj : <EOL> ( obj , tail ) = obj . split ( '<STR_LIT:.>' ) <EOL> microseconds = float ( '<STR_LIT>' + tail ) * <NUM_LIT> <EOL> hours , minutes , seconds = obj . split ( '<STR_LIT::>' ) <EOL> tdelta = datetime . timedelta ( <EOL> hours = int ( hours ) , <EOL> minutes = int ( minutes ) , <EOL> seconds = int ( seconds ) , <EOL> microseconds = int ( microseconds ) <EOL> ) <EOL> return tdelta <EOL> except ValueError : <EOL> return None <EOL> def convert_time ( obj ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> microseconds = <NUM_LIT:0> <EOL> if "<STR_LIT:.>" in obj : <EOL> ( obj , tail ) = obj . split ( '<STR_LIT:.>' ) <EOL> microseconds = float ( '<STR_LIT>' + tail ) * <NUM_LIT> <EOL> hours , minutes , seconds = obj . split ( '<STR_LIT::>' ) <EOL> return datetime . time ( hour = int ( hours ) , minute = int ( minutes ) , <EOL> second = int ( seconds ) , microsecond = int ( microseconds ) ) <EOL> except ValueError : <EOL> return None <EOL> def convert_date ( obj ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return datetime . date ( * [ int ( x ) for x in obj . split ( '<STR_LIT:->' , <NUM_LIT:2> ) ] ) <EOL> except ValueError : <EOL> return None <EOL> def convert_mysql_timestamp ( timestamp ) : <EOL> """<STR_LIT>""" <EOL> if timestamp [ <NUM_LIT:4> ] == '<STR_LIT:->' : <EOL> return convert_datetime ( timestamp ) <EOL> timestamp += "<STR_LIT:0>" * ( <NUM_LIT> - len ( timestamp ) ) <EOL> year , month , day , hour , minute , second = int ( timestamp [ : <NUM_LIT:4> ] ) , int ( timestamp [ <NUM_LIT:4> : <NUM_LIT:6> ] ) , int ( timestamp [ <NUM_LIT:6> : <NUM_LIT:8> ] ) , int ( timestamp [ <NUM_LIT:8> : <NUM_LIT:10> ] ) , int ( timestamp [ <NUM_LIT:10> : <NUM_LIT:12> ] ) , int ( timestamp [ <NUM_LIT:12> : <NUM_LIT> ] ) <EOL> try : <EOL> return datetime . datetime ( year , month , day , hour , minute , second ) <EOL> except ValueError : <EOL> return None <EOL> def convert_set ( s ) : <EOL> return set ( s . split ( "<STR_LIT:U+002C>" ) ) <EOL> def convert_bit ( b ) : <EOL> return b <EOL> def convert_characters ( connection , field , data ) : <EOL> field_charset = charset_by_id ( field . charsetnr ) . name <EOL> encoding = charset_to_encoding ( field_charset ) <EOL> if field . flags & FLAG . SET : <EOL> return convert_set ( data . decode ( encoding ) ) <EOL> if field . flags & FLAG . BINARY : <EOL> return data <EOL> if connection . use_unicode : <EOL> data = data . decode ( encoding ) <EOL> elif connection . charset != field_charset : <EOL> data = data . decode ( encoding ) <EOL> data = data . encode ( connection . encoding ) <EOL> return data <EOL> encoders = { <EOL> bool : escape_bool , <EOL> int : escape_int , <EOL> long_type : escape_int , <EOL> float : escape_float , <EOL> str : escape_str , <EOL> text_type : escape_unicode , <EOL> tuple : escape_sequence , <EOL> list : escape_sequence , <EOL> set : escape_sequence , <EOL> dict : escape_dict , <EOL> type ( None ) : escape_None , <EOL> datetime . date : escape_date , <EOL> datetime . datetime : escape_datetime , <EOL> datetime . timedelta : escape_timedelta , <EOL> datetime . time : escape_time , <EOL> time . struct_time : escape_struct_time , <EOL> Decimal : str , <EOL> } <EOL> def through ( x ) : <EOL> return x <EOL> if not PY2 or JYTHON or IRONPYTHON : <EOL> encoders [ bytes ] = escape_bytes <EOL> decoders = { <EOL> FIELD_TYPE . BIT : convert_bit , <EOL> FIELD_TYPE . TINY : int , <EOL> FIELD_TYPE . SHORT : int , <EOL> FIELD_TYPE . LONG : int , <EOL> FIELD_TYPE . FLOAT : float , <EOL> FIELD_TYPE . DOUBLE : float , <EOL> FIELD_TYPE . DECIMAL : float , <EOL> FIELD_TYPE . NEWDECIMAL : float , <EOL> FIELD_TYPE . LONGLONG : int , <EOL> FIELD_TYPE . INT24 : int , <EOL> FIELD_TYPE . YEAR : int , <EOL> FIELD_TYPE . TIMESTAMP : convert_mysql_timestamp , <EOL> FIELD_TYPE . DATETIME : convert_datetime , <EOL> FIELD_TYPE . TIME : convert_timedelta , <EOL> FIELD_TYPE . DATE : convert_date , <EOL> FIELD_TYPE . SET : convert_set , <EOL> FIELD_TYPE . BLOB : through , <EOL> FIELD_TYPE . TINY_BLOB : through , <EOL> FIELD_TYPE . MEDIUM_BLOB : through , <EOL> FIELD_TYPE . LONG_BLOB : through , <EOL> FIELD_TYPE . STRING : through , <EOL> FIELD_TYPE . VAR_STRING : through , <EOL> FIELD_TYPE . VARCHAR : through , <EOL> FIELD_TYPE . DECIMAL : Decimal , <EOL> FIELD_TYPE . NEWDECIMAL : Decimal , <EOL> } <EOL> conversions = decoders <EOL> def Thing2Literal ( obj ) : <EOL> return escape_str ( str ( obj ) ) </s>
