#NEWLINE# @ reflection . cache #NEWLINE# def get_columns( self , connection , table_name , schema = None , **kw ) : #NEWLINE# #INDENT# tblqry = "\n        SELECT r.rdb$field_name AS fname,\n                        r.rdb$null_flag AS null_flag,\n                        t.rdb$type_name AS ftype,\n                        f.rdb$field_sub_type AS stype,\n                        f.rdb$field_length/\n                            COALESCE(cs.rdb$bytes_per_character,1) AS flen,\n                        f.rdb$field_precision AS fprec,\n                        f.rdb$field_scale AS fscale,\n                        COALESCE(r.rdb$default_source,\n                                f.rdb$default_source) AS fdefault\n        FROM rdb$relation_fields r\n             JOIN rdb$fields f ON r.rdb$field_source=f.rdb$field_name\n             JOIN rdb$types t\n              ON t.rdb$type=f.rdb$field_type AND\n                    t.rdb$field_name='RDB$FIELD_TYPE'\n             LEFT JOIN rdb$character_sets cs ON\n                    f.rdb$character_set_id=cs.rdb$character_set_id\n        WHERE f.rdb$system_flag=0 AND r.rdb$relation_name=?\n        ORDER BY r.rdb$field_position\n        " #NEWLINE# pk_constraint = self . get_pk_constraint ( connection , table_name ) #NEWLINE# pkey_cols = pk_constraint [ 'constrained_columns' ] #NEWLINE# tablename = self . denormalize_name ( table_name ) #NEWLINE# c = connection . execute ( tblqry , [ tablename ] ) #NEWLINE# cols = [ ] #NEWLINE# while True : #NEWLINE# #INDENT# row = c . fetchone ( ) #NEWLINE# if ( row is None ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# name = self . normalize_name ( row [ 'fname' ] ) #NEWLINE# orig_colname = row [ 'fname' ] #NEWLINE# colspec = row [ 'ftype' ] . rstrip ( ) #NEWLINE# coltype = self . ischema_names . get ( colspec ) #NEWLINE# if ( coltype is None ) : #NEWLINE# #INDENT# util . warn ( ( "Did not recognize type '%s' of column '%s'" % ( colspec , name ) ) ) #NEWLINE# coltype = sqltypes . NULLTYPE #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( issubclass ( coltype , Integer ) and ( row [ 'fprec' ] != 0 ) ) : #NEWLINE# #INDENT# coltype = NUMERIC ( precision = row [ 'fprec' ] , scale = ( row [ 'fscale' ] * ( - 1 ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( colspec in ( 'VARYING' , 'CSTRING' ) ) : #NEWLINE# #INDENT# coltype = coltype ( row [ 'flen' ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( colspec == 'TEXT' ) : #NEWLINE# #INDENT# coltype = TEXT ( row [ 'flen' ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( colspec == 'BLOB' ) : #NEWLINE# #INDENT# if ( row [ 'stype' ] == 1 ) : #NEWLINE# #INDENT# coltype = TEXT ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# coltype = BLOB ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# coltype = coltype ( ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# defvalue = None #NEWLINE# if ( row [ 'fdefault' ] is not None ) : #NEWLINE# #INDENT# defexpr = row [ 'fdefault' ] . lstrip ( ) #NEWLINE# assert ( defexpr [ : 8 ] . rstrip ( ) . upper ( ) == 'DEFAULT' ) , ( 'Unrecognized default value: %s' % defexpr ) #NEWLINE# defvalue = defexpr [ 8 : ] . strip ( ) #NEWLINE# if ( defvalue == 'NULL' ) : #NEWLINE# #INDENT# defvalue = None #NEWLINE# #UNINDENT# #UNINDENT# col_d = { #NEWLINE# #INDENT# 'name' : name , #NEWLINE# 'type' : coltype , #NEWLINE# 'nullable' : ( not bool ( defvalue [ 'null_flag' ] ) ) , #NEWLINE# 'default' : defvalue , #NEWLINE# 'autoincrement' : ( defvalue is None ) , #NEWLINE# #UNINDENT# } #NEWLINE# if ( orig_colname . lower ( ) == orig_colname ) : #NEWLINE# #INDENT# col_d [ 'quote' ] = True #NEWLINE# #UNINDENT# if ( ( len ( pkey_cols ) == 1 ) and ( name == pkey_cols [ 0 ] ) ) : #NEWLINE# #INDENT# seq_d = self . get_column_sequence ( connection , tablename , name ) #NEWLINE# if ( seq_d is not None ) : #NEWLINE# #INDENT# col_d [ 'sequence' ] = seq_d #NEWLINE# #UNINDENT# #UNINDENT# cols . append ( col_d ) #NEWLINE# #UNINDENT# return cols