<s> """<STR_LIT>""" <EOL> import datetime <EOL> from sqlalchemy import schema as sa_schema <EOL> from sqlalchemy import exc , types as sqltypes , sql , util <EOL> from sqlalchemy . sql import expression <EOL> from sqlalchemy . engine import base , default , reflection <EOL> from sqlalchemy . sql import compiler <EOL> from sqlalchemy . types import ( BIGINT , BLOB , DATE , FLOAT , INTEGER , NUMERIC , <EOL> SMALLINT , TEXT , TIME , TIMESTAMP , Integer ) <EOL> RESERVED_WORDS = set ( [ <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:all>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:count>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:date>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:default>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:end>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:file>" , <EOL> "<STR_LIT>" , "<STR_LIT:float>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:index>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:int>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:key>" , "<STR_LIT>" , "<STR_LIT:left>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:null>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT:password>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT:right>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:size>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:start>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:time>" , "<STR_LIT>" , <EOL> "<STR_LIT:to>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:user>" , "<STR_LIT>" , "<STR_LIT:value>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> ] ) <EOL> class _StringType ( sqltypes . String ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , charset = None , ** kw ) : <EOL> self . charset = charset <EOL> super ( _StringType , self ) . __init__ ( ** kw ) <EOL> class VARCHAR ( _StringType , sqltypes . VARCHAR ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None , ** kwargs ) : <EOL> super ( VARCHAR , self ) . __init__ ( length = length , ** kwargs ) <EOL> class CHAR ( _StringType , sqltypes . CHAR ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None , ** kwargs ) : <EOL> super ( CHAR , self ) . __init__ ( length = length , ** kwargs ) <EOL> class _FBDateTime ( sqltypes . DateTime ) : <EOL> def bind_processor ( self , dialect ) : <EOL> def process ( value ) : <EOL> if type ( value ) == datetime . date : <EOL> return datetime . datetime ( value . year , value . month , value . day ) <EOL> else : <EOL> return value <EOL> return process <EOL> colspecs = { <EOL> sqltypes . DateTime : _FBDateTime <EOL> } <EOL> ischema_names = { <EOL> '<STR_LIT>' : SMALLINT , <EOL> '<STR_LIT>' : INTEGER , <EOL> '<STR_LIT>' : FLOAT , <EOL> '<STR_LIT>' : FLOAT , <EOL> '<STR_LIT>' : DATE , <EOL> '<STR_LIT>' : TIME , <EOL> '<STR_LIT>' : TEXT , <EOL> '<STR_LIT>' : BIGINT , <EOL> '<STR_LIT>' : FLOAT , <EOL> '<STR_LIT>' : TIMESTAMP , <EOL> '<STR_LIT>' : VARCHAR , <EOL> '<STR_LIT>' : CHAR , <EOL> '<STR_LIT>' : BLOB , <EOL> } <EOL> class FBTypeCompiler ( compiler . GenericTypeCompiler ) : <EOL> def visit_boolean ( self , type_ , ** kw ) : <EOL> return self . visit_SMALLINT ( type_ , ** kw ) <EOL> def visit_datetime ( self , type_ , ** kw ) : <EOL> return self . visit_TIMESTAMP ( type_ , ** kw ) <EOL> def visit_TEXT ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_BLOB ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def _extend_string ( self , type_ , basic ) : <EOL> charset = getattr ( type_ , '<STR_LIT>' , None ) <EOL> if charset is None : <EOL> return basic <EOL> else : <EOL> return '<STR_LIT>' % ( basic , charset ) <EOL> def visit_CHAR ( self , type_ , ** kw ) : <EOL> basic = super ( FBTypeCompiler , self ) . visit_CHAR ( type_ , ** kw ) <EOL> return self . _extend_string ( type_ , basic ) <EOL> def visit_VARCHAR ( self , type_ , ** kw ) : <EOL> if not type_ . length : <EOL> raise exc . CompileError ( <EOL> "<STR_LIT>" % <EOL> self . dialect . name ) <EOL> basic = super ( FBTypeCompiler , self ) . visit_VARCHAR ( type_ , ** kw ) <EOL> return self . _extend_string ( type_ , basic ) <EOL> class FBCompiler ( sql . compiler . SQLCompiler ) : <EOL> """<STR_LIT>""" <EOL> ansi_bind_rules = True <EOL> def visit_now_func ( self , fn , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_startswith_op_binary ( self , binary , operator , ** kw ) : <EOL> return '<STR_LIT>' % ( <EOL> binary . left . _compiler_dispatch ( self , ** kw ) , <EOL> binary . right . _compiler_dispatch ( self , ** kw ) ) <EOL> def visit_notstartswith_op_binary ( self , binary , operator , ** kw ) : <EOL> return '<STR_LIT>' % ( <EOL> binary . left . _compiler_dispatch ( self , ** kw ) , <EOL> binary . right . _compiler_dispatch ( self , ** kw ) ) <EOL> def visit_mod_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left , ** kw ) , <EOL> self . process ( binary . right , ** kw ) ) <EOL> def visit_alias ( self , alias , asfrom = False , ** kwargs ) : <EOL> if self . dialect . _version_two : <EOL> return super ( FBCompiler , self ) . visit_alias ( alias , asfrom = asfrom , ** kwargs ) <EOL> else : <EOL> if asfrom : <EOL> alias_name = isinstance ( alias . name , <EOL> expression . _truncated_label ) and self . _truncated_identifier ( "<STR_LIT>" , <EOL> alias . name ) or alias . name <EOL> return self . process ( <EOL> alias . original , asfrom = asfrom , ** kwargs ) + "<STR_LIT:U+0020>" + self . preparer . format_alias ( alias , alias_name ) <EOL> else : <EOL> return self . process ( alias . original , ** kwargs ) <EOL> def visit_substring_func ( self , func , ** kw ) : <EOL> s = self . process ( func . clauses . clauses [ <NUM_LIT:0> ] ) <EOL> start = self . process ( func . clauses . clauses [ <NUM_LIT:1> ] ) <EOL> if len ( func . clauses . clauses ) > <NUM_LIT:2> : <EOL> length = self . process ( func . clauses . clauses [ <NUM_LIT:2> ] ) <EOL> return "<STR_LIT>" % ( s , start , length ) <EOL> else : <EOL> return "<STR_LIT>" % ( s , start ) <EOL> def visit_length_func ( self , function , ** kw ) : <EOL> if self . dialect . _version_two : <EOL> return "<STR_LIT>" + self . function_argspec ( function ) <EOL> else : <EOL> return "<STR_LIT>" + self . function_argspec ( function ) <EOL> visit_char_length_func = visit_length_func <EOL> def function_argspec ( self , func , ** kw ) : <EOL> if func . clauses is not None and len ( func . clauses ) : <EOL> return self . process ( func . clause_expr , ** kw ) <EOL> else : <EOL> return "<STR_LIT>" <EOL> def default_from ( self ) : <EOL> return "<STR_LIT>" <EOL> def visit_sequence ( self , seq ) : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( seq ) <EOL> def get_select_precolumns ( self , select , ** kw ) : <EOL> """<STR_LIT>""" <EOL> result = "<STR_LIT>" <EOL> if select . _limit_clause is not None : <EOL> result += "<STR_LIT>" % self . process ( select . _limit_clause , ** kw ) <EOL> if select . _offset_clause is not None : <EOL> result += "<STR_LIT>" % self . process ( select . _offset_clause , ** kw ) <EOL> if select . _distinct : <EOL> result += "<STR_LIT>" <EOL> return result <EOL> def limit_clause ( self , select , ** kw ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" <EOL> def returning_clause ( self , stmt , returning_cols ) : <EOL> columns = [ <EOL> self . _label_select_column ( None , c , True , False , { } ) <EOL> for c in expression . _select_iterables ( returning_cols ) <EOL> ] <EOL> return '<STR_LIT>' + '<STR_LIT:U+002CU+0020>' . join ( columns ) <EOL> class FBDDLCompiler ( sql . compiler . DDLCompiler ) : <EOL> """<STR_LIT>""" <EOL> def visit_create_sequence ( self , create ) : <EOL> """<STR_LIT>""" <EOL> if create . element . start is not None : <EOL> raise NotImplemented ( <EOL> "<STR_LIT>" ) <EOL> if create . element . increment is not None : <EOL> raise NotImplemented ( <EOL> "<STR_LIT>" ) <EOL> if self . dialect . _version_two : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( create . element ) <EOL> else : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( create . element ) <EOL> def visit_drop_sequence ( self , drop ) : <EOL> """<STR_LIT>""" <EOL> if self . dialect . _version_two : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( drop . element ) <EOL> else : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( drop . element ) <EOL> class FBIdentifierPreparer ( sql . compiler . IdentifierPreparer ) : <EOL> """<STR_LIT>""" <EOL> reserved_words = RESERVED_WORDS <EOL> illegal_initial_characters = compiler . ILLEGAL_INITIAL_CHARACTERS . union ( <EOL> [ '<STR_LIT:_>' ] ) <EOL> def __init__ ( self , dialect ) : <EOL> super ( FBIdentifierPreparer , self ) . __init__ ( dialect , omit_schema = True ) <EOL> class FBExecutionContext ( default . DefaultExecutionContext ) : <EOL> def fire_sequence ( self , seq , type_ ) : <EOL> """<STR_LIT>""" <EOL> return self . _execute_scalar ( <EOL> "<STR_LIT>" % <EOL> self . dialect . identifier_preparer . format_sequence ( seq ) , <EOL> type_ <EOL> ) <EOL> class FBDialect ( default . DefaultDialect ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> max_identifier_length = <NUM_LIT> <EOL> supports_sequences = True <EOL> sequences_optional = False <EOL> supports_default_values = True <EOL> postfetch_lastrowid = False <EOL> supports_native_boolean = False <EOL> requires_name_normalize = True <EOL> supports_empty_insert = False <EOL> statement_compiler = FBCompiler <EOL> ddl_compiler = FBDDLCompiler <EOL> preparer = FBIdentifierPreparer <EOL> type_compiler = FBTypeCompiler <EOL> execution_ctx_cls = FBExecutionContext <EOL> colspecs = colspecs <EOL> ischema_names = ischema_names <EOL> construct_arguments = [ ] <EOL> _version_two = True <EOL> def initialize ( self , connection ) : <EOL> super ( FBDialect , self ) . initialize ( connection ) <EOL> self . _version_two = ( '<STR_LIT>' in self . server_version_info and <EOL> self . server_version_info >= ( <NUM_LIT:2> , ) <EOL> ) or ( '<STR_LIT>' in self . server_version_info and <EOL> self . server_version_info >= ( <NUM_LIT:6> , ) <EOL> ) <EOL> if not self . _version_two : <EOL> self . ischema_names = ischema_names . copy ( ) <EOL> self . ischema_names [ '<STR_LIT>' ] = sqltypes . DATE <EOL> self . colspecs = { <EOL> sqltypes . DateTime : sqltypes . DATE <EOL> } <EOL> self . implicit_returning = self . _version_two and self . __dict__ . get ( '<STR_LIT>' , True ) <EOL> def normalize_name ( self , name ) : <EOL> name = name and name . rstrip ( ) <EOL> if name is None : <EOL> return None <EOL> elif name . upper ( ) == name and not self . identifier_preparer . _requires_quotes ( name . lower ( ) ) : <EOL> return name . lower ( ) <EOL> else : <EOL> return name <EOL> def denormalize_name ( self , name ) : <EOL> if name is None : <EOL> return None <EOL> elif name . lower ( ) == name and not self . identifier_preparer . _requires_quotes ( name . lower ( ) ) : <EOL> return name . upper ( ) <EOL> else : <EOL> return name <EOL> def has_table ( self , connection , table_name , schema = None ) : <EOL> """<STR_LIT>""" <EOL> tblqry = """<STR_LIT>""" <EOL> c = connection . execute ( tblqry , [ self . denormalize_name ( table_name ) ] ) <EOL> return c . first ( ) is not None <EOL> def has_sequence ( self , connection , sequence_name , schema = None ) : <EOL> """<STR_LIT>""" <EOL> genqry = """<STR_LIT>""" <EOL> c = connection . execute ( genqry , [ self . denormalize_name ( sequence_name ) ] ) <EOL> return c . first ( ) is not None <EOL> @ reflection . cache <EOL> def get_table_names ( self , connection , schema = None , ** kw ) : <EOL> s = """<STR_LIT>""" <EOL> return [ self . normalize_name ( row [ <NUM_LIT:0> ] ) for row in connection . execute ( s ) ] <EOL> @ reflection . cache <EOL> def get_view_names ( self , connection , schema = None , ** kw ) : <EOL> s = """<STR_LIT>""" <EOL> return [ self . normalize_name ( row [ <NUM_LIT:0> ] ) for row in connection . execute ( s ) ] <EOL> @ reflection . cache <EOL> def get_view_definition ( self , connection , view_name , schema = None , ** kw ) : <EOL> qry = """<STR_LIT>""" <EOL> rp = connection . execute ( qry , [ self . denormalize_name ( view_name ) ] ) <EOL> row = rp . first ( ) <EOL> if row : <EOL> return row [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> @ reflection . cache <EOL> def get_pk_constraint ( self , connection , table_name , schema = None , ** kw ) : <EOL> keyqry = """<STR_LIT>""" <EOL> tablename = self . denormalize_name ( table_name ) <EOL> c = connection . execute ( keyqry , [ "<STR_LIT>" , tablename ] ) <EOL> pkfields = [ self . normalize_name ( r [ '<STR_LIT>' ] ) for r in c . fetchall ( ) ] <EOL> return { '<STR_LIT>' : pkfields , '<STR_LIT:name>' : None } <EOL> @ reflection . cache <EOL> def get_column_sequence ( self , connection , <EOL> table_name , column_name , <EOL> schema = None , ** kw ) : <EOL> tablename = self . denormalize_name ( table_name ) <EOL> colname = self . denormalize_name ( column_name ) <EOL> genqry = """<STR_LIT>""" <EOL> genr = connection . execute ( genqry , [ tablename , colname ] ) . first ( ) <EOL> if genr is not None : <EOL> return dict ( name = self . normalize_name ( genr [ '<STR_LIT>' ] ) ) <EOL> @ reflection . cache <EOL> def get_columns ( self , connection , table_name , schema = None , ** kw ) : <EOL> tblqry = """<STR_LIT>""" <EOL> pk_constraint = self . get_pk_constraint ( connection , table_name ) <EOL> pkey_cols = pk_constraint [ '<STR_LIT>' ] <EOL> tablename = self . denormalize_name ( table_name ) <EOL> c = connection . execute ( tblqry , [ tablename ] ) <EOL> cols = [ ] <EOL> while True : <EOL> row = c . fetchone ( ) <EOL> if row is None : <EOL> break <EOL> name = self . normalize_name ( row [ '<STR_LIT>' ] ) <EOL> orig_colname = row [ '<STR_LIT>' ] <EOL> colspec = row [ '<STR_LIT>' ] . rstrip ( ) <EOL> coltype = self . ischema_names . get ( colspec ) <EOL> if coltype is None : <EOL> util . warn ( "<STR_LIT>" % <EOL> ( colspec , name ) ) <EOL> coltype = sqltypes . NULLTYPE <EOL> elif issubclass ( coltype , Integer ) and row [ '<STR_LIT>' ] != <NUM_LIT:0> : <EOL> coltype = NUMERIC ( <EOL> precision = row [ '<STR_LIT>' ] , <EOL> scale = row [ '<STR_LIT>' ] * - <NUM_LIT:1> ) <EOL> elif colspec in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> coltype = coltype ( row [ '<STR_LIT>' ] ) <EOL> elif colspec == '<STR_LIT>' : <EOL> coltype = TEXT ( row [ '<STR_LIT>' ] ) <EOL> elif colspec == '<STR_LIT>' : <EOL> if row [ '<STR_LIT>' ] == <NUM_LIT:1> : <EOL> coltype = TEXT ( ) <EOL> else : <EOL> coltype = BLOB ( ) <EOL> else : <EOL> coltype = coltype ( ) <EOL> defvalue = None <EOL> if row [ '<STR_LIT>' ] is not None : <EOL> defexpr = row [ '<STR_LIT>' ] . lstrip ( ) <EOL> assert defexpr [ : <NUM_LIT:8> ] . rstrip ( ) . upper ( ) == '<STR_LIT>' , "<STR_LIT>" % defexpr <EOL> defvalue = defexpr [ <NUM_LIT:8> : ] . strip ( ) <EOL> if defvalue == '<STR_LIT>' : <EOL> defvalue = None <EOL> col_d = { <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT:type>' : coltype , <EOL> '<STR_LIT>' : not bool ( row [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT:default>' : defvalue , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> if orig_colname . lower ( ) == orig_colname : <EOL> col_d [ '<STR_LIT>' ] = True <EOL> if len ( pkey_cols ) == <NUM_LIT:1> and name == pkey_cols [ <NUM_LIT:0> ] : <EOL> seq_d = self . get_column_sequence ( connection , tablename , name ) <EOL> if seq_d is not None : <EOL> col_d [ '<STR_LIT>' ] = seq_d <EOL> cols . append ( col_d ) <EOL> return cols <EOL> @ reflection . cache <EOL> def get_foreign_keys ( self , connection , table_name , schema = None , ** kw ) : <EOL> fkqry = """<STR_LIT>""" <EOL> tablename = self . denormalize_name ( table_name ) <EOL> c = connection . execute ( fkqry , [ "<STR_LIT>" , tablename ] ) <EOL> fks = util . defaultdict ( lambda : { <EOL> '<STR_LIT:name>' : None , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : [ ] <EOL> } ) <EOL> for row in c : <EOL> cname = self . normalize_name ( row [ '<STR_LIT>' ] ) <EOL> fk = fks [ cname ] <EOL> if not fk [ '<STR_LIT:name>' ] : <EOL> fk [ '<STR_LIT:name>' ] = cname <EOL> fk [ '<STR_LIT>' ] = self . normalize_name ( row [ '<STR_LIT>' ] ) <EOL> fk [ '<STR_LIT>' ] . append ( <EOL> self . normalize_name ( row [ '<STR_LIT>' ] ) ) <EOL> fk [ '<STR_LIT>' ] . append ( <EOL> self . normalize_name ( row [ '<STR_LIT>' ] ) ) <EOL> return list ( fks . values ( ) ) <EOL> @ reflection . cache <EOL> def get_indexes ( self , connection , table_name , schema = None , ** kw ) : <EOL> qry = """<STR_LIT>""" <EOL> c = connection . execute ( qry , [ self . denormalize_name ( table_name ) ] ) <EOL> indexes = util . defaultdict ( dict ) <EOL> for row in c : <EOL> indexrec = indexes [ row [ '<STR_LIT>' ] ] <EOL> if '<STR_LIT:name>' not in indexrec : <EOL> indexrec [ '<STR_LIT:name>' ] = self . normalize_name ( row [ '<STR_LIT>' ] ) <EOL> indexrec [ '<STR_LIT>' ] = [ ] <EOL> indexrec [ '<STR_LIT>' ] = bool ( row [ '<STR_LIT>' ] ) <EOL> indexrec [ '<STR_LIT>' ] . append ( <EOL> self . normalize_name ( row [ '<STR_LIT>' ] ) ) <EOL> return list ( indexes . values ( ) ) </s>
