<s> import sqlalchemy <EOL> import csv <EOL> import codecs <EOL> from cubes import compat <EOL> class UTF8Recoder : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , encoding ) : <EOL> assert '<STR_LIT:b>' in f . mode , "<STR_LIT>" <EOL> self . reader = codecs . getreader ( encoding ) ( f ) <EOL> self . next = self . __next__ <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def __next__ ( self ) : <EOL> return compat . to_unicode ( next ( self . reader ) ) <EOL> class UnicodeReader : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , dialect = csv . excel , encoding = "<STR_LIT:utf-8>" , ** kwds ) : <EOL> f = UTF8Recoder ( f , encoding ) <EOL> self . reader = csv . reader ( f , dialect = dialect , ** kwds ) <EOL> self . next = self . __next__ <EOL> def __next__ ( self ) : <EOL> row = next ( self . reader ) <EOL> return [ compat . to_unicode ( s ) for s in row ] <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def create_table_from_csv ( connectable , file_name , table_name , fields , <EOL> create_id = False , schema = None ) : <EOL> """<STR_LIT>""" <EOL> metadata = sqlalchemy . MetaData ( bind = connectable ) <EOL> table = sqlalchemy . Table ( table_name , metadata , autoload = False , schema = schema ) <EOL> if table . exists ( ) : <EOL> table . drop ( checkfirst = False ) <EOL> type_map = { "<STR_LIT>" : sqlalchemy . Integer , <EOL> "<STR_LIT:float>" : sqlalchemy . Numeric , <EOL> "<STR_LIT:string>" : sqlalchemy . String ( <NUM_LIT> ) , <EOL> "<STR_LIT:text>" : sqlalchemy . Text , <EOL> "<STR_LIT:date>" : sqlalchemy . Text , <EOL> "<STR_LIT>" : sqlalchemy . Integer } <EOL> if create_id : <EOL> col = sqlalchemy . schema . Column ( '<STR_LIT:id>' , sqlalchemy . Integer , primary_key = True ) <EOL> table . append_column ( col ) <EOL> field_names = [ ] <EOL> for ( field_name , field_type ) in fields : <EOL> col = sqlalchemy . schema . Column ( field_name , type_map [ field_type . lower ( ) ] ) <EOL> table . append_column ( col ) <EOL> field_names . append ( field_name ) <EOL> table . create ( ) <EOL> reader = UnicodeReader ( open ( file_name , '<STR_LIT:rb>' ) ) <EOL> next ( reader ) <EOL> insert_command = table . insert ( ) <EOL> for row in reader : <EOL> record = dict ( zip ( field_names , row ) ) <EOL> insert_command . execute ( record ) </s>
