<s> """<STR_LIT>""" <EOL> from collections import defaultdict <EOL> import re <EOL> from ... import sql , schema , exc , util <EOL> from ... engine import default , reflection <EOL> from ... sql import compiler , expression , operators <EOL> from ... import types as sqltypes <EOL> try : <EOL> from uuid import UUID as _python_UUID <EOL> except ImportError : <EOL> _python_UUID = None <EOL> from sqlalchemy . types import INTEGER , BIGINT , SMALLINT , VARCHAR , CHAR , TEXT , FLOAT , NUMERIC , DATE , BOOLEAN , REAL <EOL> RESERVED_WORDS = set ( <EOL> [ "<STR_LIT:all>" , "<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>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT:default>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:end>" , <EOL> "<STR_LIT>" , "<STR_LIT:false>" , "<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>" , <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>" , "<STR_LIT>" , "<STR_LIT:to>" , "<STR_LIT>" , "<STR_LIT:true>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:user>" , <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>" , "<STR_LIT:left>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:right>" , "<STR_LIT>" , "<STR_LIT>" <EOL> ] ) <EOL> _DECIMAL_TYPES = ( <NUM_LIT> , <NUM_LIT> ) <EOL> _FLOAT_TYPES = ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> _INT_TYPES = ( <NUM_LIT:20> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> class BYTEA ( sqltypes . LargeBinary ) : <EOL> __visit_name__ = '<STR_LIT>' <EOL> class DOUBLE_PRECISION ( sqltypes . Float ) : <EOL> __visit_name__ = '<STR_LIT>' <EOL> class INET ( sqltypes . TypeEngine ) : <EOL> __visit_name__ = "<STR_LIT>" <EOL> PGInet = INET <EOL> class CIDR ( sqltypes . TypeEngine ) : <EOL> __visit_name__ = "<STR_LIT>" <EOL> PGCidr = CIDR <EOL> class MACADDR ( sqltypes . TypeEngine ) : <EOL> __visit_name__ = "<STR_LIT>" <EOL> PGMacAddr = MACADDR <EOL> class OID ( sqltypes . TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = "<STR_LIT>" <EOL> class TIMESTAMP ( sqltypes . TIMESTAMP ) : <EOL> def __init__ ( self , timezone = False , precision = None ) : <EOL> super ( TIMESTAMP , self ) . __init__ ( timezone = timezone ) <EOL> self . precision = precision <EOL> class TIME ( sqltypes . TIME ) : <EOL> def __init__ ( self , timezone = False , precision = None ) : <EOL> super ( TIME , self ) . __init__ ( timezone = timezone ) <EOL> self . precision = precision <EOL> class INTERVAL ( sqltypes . TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , precision = None ) : <EOL> self . precision = precision <EOL> @ classmethod <EOL> def _adapt_from_generic_interval ( cls , interval ) : <EOL> return INTERVAL ( precision = interval . second_precision ) <EOL> @ property <EOL> def _type_affinity ( self ) : <EOL> return sqltypes . Interval <EOL> PGInterval = INTERVAL <EOL> class BIT ( sqltypes . TypeEngine ) : <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , length = None , varying = False ) : <EOL> if not varying : <EOL> self . length = length or <NUM_LIT:1> <EOL> else : <EOL> self . length = length <EOL> self . varying = varying <EOL> PGBit = BIT <EOL> class UUID ( sqltypes . TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , as_uuid = False ) : <EOL> """<STR_LIT>""" <EOL> if as_uuid and _python_UUID is None : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> self . as_uuid = as_uuid <EOL> def bind_processor ( self , dialect ) : <EOL> if self . as_uuid : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = util . text_type ( value ) <EOL> return value <EOL> return process <EOL> else : <EOL> return None <EOL> def result_processor ( self , dialect , coltype ) : <EOL> if self . as_uuid : <EOL> def process ( value ) : <EOL> if value is not None : <EOL> value = _python_UUID ( value ) <EOL> return value <EOL> return process <EOL> else : <EOL> return None <EOL> PGUuid = UUID <EOL> class TSVECTOR ( sqltypes . TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class _Slice ( expression . ColumnElement ) : <EOL> __visit_name__ = '<STR_LIT>' <EOL> type = sqltypes . NULLTYPE <EOL> def __init__ ( self , slice_ , source_comparator ) : <EOL> self . start = source_comparator . _check_literal ( <EOL> source_comparator . expr , <EOL> operators . getitem , slice_ . start ) <EOL> self . stop = source_comparator . _check_literal ( <EOL> source_comparator . expr , <EOL> operators . getitem , slice_ . stop ) <EOL> class Any ( expression . ColumnElement ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , left , right , operator = operators . eq ) : <EOL> self . type = sqltypes . Boolean ( ) <EOL> self . left = expression . _literal_as_binds ( left ) <EOL> self . right = right <EOL> self . operator = operator <EOL> class All ( expression . ColumnElement ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT:all>' <EOL> def __init__ ( self , left , right , operator = operators . eq ) : <EOL> self . type = sqltypes . Boolean ( ) <EOL> self . left = expression . _literal_as_binds ( left ) <EOL> self . right = right <EOL> self . operator = operator <EOL> class array ( expression . Tuple ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> def __init__ ( self , clauses , ** kw ) : <EOL> super ( array , self ) . __init__ ( * clauses , ** kw ) <EOL> self . type = ARRAY ( self . type ) <EOL> def _bind_param ( self , operator , obj ) : <EOL> return array ( [ <EOL> expression . BindParameter ( None , o , _compared_to_operator = operator , <EOL> _compared_to_type = self . type , unique = True ) <EOL> for o in obj <EOL> ] ) <EOL> def self_group ( self , against = None ) : <EOL> return self <EOL> class ARRAY ( sqltypes . Concatenable , sqltypes . TypeEngine ) : <EOL> """<STR_LIT>""" <EOL> __visit_name__ = '<STR_LIT>' <EOL> class Comparator ( sqltypes . Concatenable . Comparator ) : <EOL> """<STR_LIT>""" <EOL> def __getitem__ ( self , index ) : <EOL> shift_indexes = <NUM_LIT:1> if self . expr . type . zero_indexes else <NUM_LIT:0> <EOL> if isinstance ( index , slice ) : <EOL> if shift_indexes : <EOL> index = slice ( <EOL> index . start + shift_indexes , <EOL> index . stop + shift_indexes , <EOL> index . step <EOL> ) <EOL> index = _Slice ( index , self ) <EOL> return_type = self . type <EOL> else : <EOL> index += shift_indexes <EOL> return_type = self . type . item_type <EOL> return self . _binary_operate ( self . expr , operators . getitem , index , <EOL> result_type = return_type ) <EOL> def any ( self , other , operator = operators . eq ) : <EOL> """<STR_LIT>""" <EOL> return Any ( other , self . expr , operator = operator ) <EOL> def all ( self , other , operator = operators . eq ) : <EOL> """<STR_LIT>""" <EOL> return All ( other , self . expr , operator = operator ) <EOL> def contains ( self , other , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . expr . op ( '<STR_LIT>' ) ( other ) <EOL> def contained_by ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . expr . op ( '<STR_LIT>' ) ( other ) <EOL> def overlap ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . expr . op ( '<STR_LIT>' ) ( other ) <EOL> def _adapt_expression ( self , op , other_comparator ) : <EOL> if isinstance ( op , operators . custom_op ) : <EOL> if op . opstring in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return op , sqltypes . Boolean <EOL> return sqltypes . Concatenable . Comparator . _adapt_expression ( self , op , other_comparator ) <EOL> comparator_factory = Comparator <EOL> def __init__ ( self , item_type , as_tuple = False , dimensions = None , <EOL> zero_indexes = False ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( item_type , ARRAY ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if isinstance ( item_type , type ) : <EOL> item_type = item_type ( ) <EOL> self . item_type = item_type <EOL> self . as_tuple = as_tuple <EOL> self . dimensions = dimensions <EOL> self . zero_indexes = zero_indexes <EOL> @ property <EOL> def python_type ( self ) : <EOL> return list <EOL> def compare_values ( self , x , y ) : <EOL> return x == y <EOL> def _proc_array ( self , arr , itemproc , dim , collection ) : <EOL> if dim is None : <EOL> arr = list ( arr ) <EOL> if dim == <NUM_LIT:1> or dim is None and ( <EOL> not arr or not isinstance ( arr [ <NUM_LIT:0> ] , ( list , tuple ) ) ) : <EOL> if itemproc : <EOL> return collection ( itemproc ( x ) for x in arr ) <EOL> else : <EOL> return collection ( arr ) <EOL> else : <EOL> return collection ( <EOL> self . _proc_array ( <EOL> x , itemproc , <EOL> dim - <NUM_LIT:1> if dim is not None else None , <EOL> collection ) <EOL> for x in arr <EOL> ) <EOL> def bind_processor ( self , dialect ) : <EOL> item_proc = self . item_type . dialect_impl ( dialect ) . bind_processor ( dialect ) <EOL> def process ( value ) : <EOL> if value is None : <EOL> return value <EOL> else : <EOL> return self . _proc_array ( <EOL> value , <EOL> item_proc , <EOL> self . dimensions , <EOL> list ) <EOL> return process <EOL> def result_processor ( self , dialect , coltype ) : <EOL> item_proc = self . item_type . dialect_impl ( dialect ) . result_processor ( dialect , coltype ) <EOL> def process ( value ) : <EOL> if value is None : <EOL> return value <EOL> else : <EOL> return self . _proc_array ( <EOL> value , <EOL> item_proc , <EOL> self . dimensions , <EOL> tuple if self . as_tuple else list ) <EOL> return process <EOL> PGArray = ARRAY <EOL> class ENUM ( sqltypes . Enum ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * enums , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . create_type = kw . pop ( "<STR_LIT>" , True ) <EOL> super ( ENUM , self ) . __init__ ( * enums , ** kw ) <EOL> def create ( self , bind = None , checkfirst = True ) : <EOL> """<STR_LIT>""" <EOL> if not bind . dialect . supports_native_enum : <EOL> return <EOL> if not checkfirst or not bind . dialect . has_type ( <EOL> bind , self . name , schema = self . schema ) : <EOL> bind . execute ( CreateEnumType ( self ) ) <EOL> def drop ( self , bind = None , checkfirst = True ) : <EOL> """<STR_LIT>""" <EOL> if not bind . dialect . supports_native_enum : <EOL> return <EOL> if not checkfirst or bind . dialect . has_type ( bind , self . name , schema = self . schema ) : <EOL> bind . execute ( DropEnumType ( self ) ) <EOL> def _check_for_name_in_memos ( self , checkfirst , kw ) : <EOL> """<STR_LIT>""" <EOL> if not self . create_type : <EOL> return True <EOL> if '<STR_LIT>' in kw : <EOL> ddl_runner = kw [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in ddl_runner . memo : <EOL> pg_enums = ddl_runner . memo [ '<STR_LIT>' ] <EOL> else : <EOL> pg_enums = ddl_runner . memo [ '<STR_LIT>' ] = set ( ) <EOL> present = self . name in pg_enums <EOL> pg_enums . add ( self . name ) <EOL> return present <EOL> else : <EOL> return False <EOL> def _on_table_create ( self , target , bind , checkfirst , ** kw ) : <EOL> if not self . _check_for_name_in_memos ( checkfirst , kw ) : <EOL> self . create ( bind = bind , checkfirst = checkfirst ) <EOL> def _on_metadata_create ( self , target , bind , checkfirst , ** kw ) : <EOL> if not self . _check_for_name_in_memos ( checkfirst , kw ) : <EOL> self . create ( bind = bind , checkfirst = checkfirst ) <EOL> def _on_metadata_drop ( self , target , bind , checkfirst , ** kw ) : <EOL> if not self . _check_for_name_in_memos ( checkfirst , kw ) : <EOL> self . drop ( bind = bind , checkfirst = checkfirst ) <EOL> colspecs = { <EOL> sqltypes . Interval : INTERVAL , <EOL> sqltypes . Enum : ENUM , <EOL> } <EOL> ischema_names = { <EOL> '<STR_LIT>' : INTEGER , <EOL> '<STR_LIT>' : BIGINT , <EOL> '<STR_LIT>' : SMALLINT , <EOL> '<STR_LIT>' : VARCHAR , <EOL> '<STR_LIT>' : CHAR , <EOL> '<STR_LIT>' : sqltypes . String , <EOL> '<STR_LIT:name>' : sqltypes . String , <EOL> '<STR_LIT:text>' : TEXT , <EOL> '<STR_LIT>' : NUMERIC , <EOL> '<STR_LIT:float>' : FLOAT , <EOL> '<STR_LIT>' : REAL , <EOL> '<STR_LIT>' : INET , <EOL> '<STR_LIT>' : CIDR , <EOL> '<STR_LIT>' : UUID , <EOL> '<STR_LIT>' : BIT , <EOL> '<STR_LIT>' : BIT , <EOL> '<STR_LIT>' : MACADDR , <EOL> '<STR_LIT>' : OID , <EOL> '<STR_LIT>' : DOUBLE_PRECISION , <EOL> '<STR_LIT>' : TIMESTAMP , <EOL> '<STR_LIT>' : TIMESTAMP , <EOL> '<STR_LIT>' : TIMESTAMP , <EOL> '<STR_LIT>' : TIME , <EOL> '<STR_LIT>' : TIME , <EOL> '<STR_LIT:date>' : DATE , <EOL> '<STR_LIT:time>' : TIME , <EOL> '<STR_LIT>' : BYTEA , <EOL> '<STR_LIT>' : BOOLEAN , <EOL> '<STR_LIT>' : INTERVAL , <EOL> '<STR_LIT>' : INTERVAL , <EOL> '<STR_LIT>' : INTERVAL , <EOL> '<STR_LIT>' : TSVECTOR <EOL> } <EOL> class PGCompiler ( compiler . SQLCompiler ) : <EOL> def visit_array ( self , element , ** kw ) : <EOL> return "<STR_LIT>" % self . visit_clauselist ( element , ** kw ) <EOL> def visit_slice ( self , element , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( element . start , ** kw ) , <EOL> self . process ( element . stop , ** kw ) , <EOL> ) <EOL> def visit_any ( self , element , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( element . left , ** kw ) , <EOL> compiler . OPERATORS [ element . operator ] , <EOL> self . process ( element . right , ** kw ) <EOL> ) <EOL> def visit_all ( self , element , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( element . left , ** kw ) , <EOL> compiler . OPERATORS [ element . operator ] , <EOL> self . process ( element . right , ** kw ) <EOL> ) <EOL> def visit_getitem_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left , ** kw ) , <EOL> self . process ( binary . right , ** kw ) <EOL> ) <EOL> def visit_match_op_binary ( self , binary , operator , ** kw ) : <EOL> if "<STR_LIT>" in binary . modifiers : <EOL> regconfig = self . render_literal_value ( <EOL> binary . modifiers [ '<STR_LIT>' ] , <EOL> sqltypes . STRINGTYPE ) <EOL> if regconfig : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left , ** kw ) , <EOL> regconfig , <EOL> self . process ( binary . right , ** kw ) <EOL> ) <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left , ** kw ) , <EOL> self . process ( binary . right , ** kw ) <EOL> ) <EOL> def visit_ilike_op_binary ( self , binary , operator , ** kw ) : <EOL> escape = binary . modifiers . get ( "<STR_LIT>" , None ) <EOL> return '<STR_LIT>' % ( self . process ( binary . left , ** kw ) , <EOL> self . process ( binary . right , ** kw ) ) + ( <EOL> '<STR_LIT>' + <EOL> self . render_literal_value ( escape , sqltypes . STRINGTYPE ) <EOL> if escape else '<STR_LIT>' <EOL> ) <EOL> def visit_notilike_op_binary ( self , binary , operator , ** kw ) : <EOL> escape = binary . modifiers . get ( "<STR_LIT>" , None ) <EOL> return '<STR_LIT>' % ( self . process ( binary . left , ** kw ) , <EOL> self . process ( binary . right , ** kw ) ) + ( <EOL> '<STR_LIT>' + <EOL> self . render_literal_value ( escape , sqltypes . STRINGTYPE ) <EOL> if escape else '<STR_LIT>' <EOL> ) <EOL> def render_literal_value ( self , value , type_ ) : <EOL> value = super ( PGCompiler , self ) . render_literal_value ( value , type_ ) <EOL> if self . dialect . _backslash_escapes : <EOL> value = value . replace ( '<STR_LIT:\\>' , '<STR_LIT>' ) <EOL> return value <EOL> def visit_sequence ( self , seq ) : <EOL> return "<STR_LIT>" % self . preparer . format_sequence ( seq ) <EOL> def limit_clause ( self , select ) : <EOL> text = "<STR_LIT>" <EOL> if select . _limit is not None : <EOL> text += "<STR_LIT>" + self . process ( sql . literal ( select . _limit ) ) <EOL> if select . _offset is not None : <EOL> if select . _limit is None : <EOL> text += "<STR_LIT>" <EOL> text += "<STR_LIT>" + self . process ( sql . literal ( select . _offset ) ) <EOL> return text <EOL> def format_from_hint_text ( self , sqltext , table , hint , iscrud ) : <EOL> if hint . upper ( ) != '<STR_LIT>' : <EOL> raise exc . CompileError ( "<STR_LIT>" % hint ) <EOL> return "<STR_LIT>" + sqltext <EOL> def get_select_precolumns ( self , select ) : <EOL> if select . _distinct is not False : <EOL> if select . _distinct is True : <EOL> return "<STR_LIT>" <EOL> elif isinstance ( select . _distinct , ( list , tuple ) ) : <EOL> return "<STR_LIT>" + '<STR_LIT:U+002CU+0020>' . join ( <EOL> [ self . process ( col ) for col in select . _distinct ] <EOL> ) + "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" + self . process ( select . _distinct ) + "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" <EOL> def for_update_clause ( self , select ) : <EOL> if select . _for_update_arg . read : <EOL> tmp = "<STR_LIT>" <EOL> else : <EOL> tmp = "<STR_LIT>" <EOL> if select . _for_update_arg . of : <EOL> tables = util . OrderedSet ( <EOL> c . table if isinstance ( c , expression . ColumnClause ) <EOL> else c for c in select . _for_update_arg . of ) <EOL> tmp += "<STR_LIT>" + "<STR_LIT:U+002CU+0020>" . join ( <EOL> self . process ( table , ashint = True ) <EOL> for table in tables <EOL> ) <EOL> if select . _for_update_arg . nowait : <EOL> tmp += "<STR_LIT>" <EOL> return tmp <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> def visit_substring_func ( self , func , ** kw ) : <EOL> s = self . process ( func . clauses . clauses [ <NUM_LIT:0> ] , ** kw ) <EOL> start = self . process ( func . clauses . clauses [ <NUM_LIT:1> ] , ** kw ) <EOL> if len ( func . clauses . clauses ) > <NUM_LIT:2> : <EOL> length = self . process ( func . clauses . clauses [ <NUM_LIT:2> ] , ** kw ) <EOL> return "<STR_LIT>" % ( s , start , length ) <EOL> else : <EOL> return "<STR_LIT>" % ( s , start ) <EOL> class PGDDLCompiler ( compiler . DDLCompiler ) : <EOL> def get_column_specification ( self , column , ** kwargs ) : <EOL> colspec = self . preparer . format_column ( column ) <EOL> impl_type = column . type . dialect_impl ( self . dialect ) <EOL> if column . primary_key and column is column . table . _autoincrement_column and ( <EOL> self . dialect . supports_smallserial or <EOL> not isinstance ( impl_type , sqltypes . SmallInteger ) <EOL> ) and ( <EOL> column . default is None or <EOL> ( <EOL> isinstance ( column . default , schema . Sequence ) and <EOL> column . default . optional <EOL> ) ) : <EOL> if isinstance ( impl_type , sqltypes . BigInteger ) : <EOL> colspec += "<STR_LIT>" <EOL> elif isinstance ( impl_type , sqltypes . SmallInteger ) : <EOL> colspec += "<STR_LIT>" <EOL> else : <EOL> colspec += "<STR_LIT>" <EOL> else : <EOL> colspec += "<STR_LIT:U+0020>" + self . dialect . type_compiler . process ( column . type ) <EOL> default = self . get_column_default_string ( column ) <EOL> if default is not None : <EOL> colspec += "<STR_LIT>" + default <EOL> if not column . nullable : <EOL> colspec += "<STR_LIT>" <EOL> return colspec <EOL> def visit_create_enum_type ( self , create ) : <EOL> type_ = create . element <EOL> return "<STR_LIT>" % ( <EOL> self . preparer . format_type ( type_ ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( <EOL> self . sql_compiler . process ( sql . literal ( e ) , literal_binds = True ) <EOL> for e in type_ . enums ) <EOL> ) <EOL> def visit_drop_enum_type ( self , drop ) : <EOL> type_ = drop . element <EOL> return "<STR_LIT>" % ( <EOL> self . preparer . format_type ( type_ ) <EOL> ) <EOL> def visit_create_index ( self , create ) : <EOL> preparer = self . preparer <EOL> index = create . element <EOL> self . _verify_index_table ( index ) <EOL> text = "<STR_LIT>" <EOL> if index . unique : <EOL> text += "<STR_LIT>" <EOL> text += "<STR_LIT>" <EOL> concurrently = index . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if concurrently : <EOL> text += "<STR_LIT>" <EOL> text += "<STR_LIT>" % ( <EOL> self . _prepared_index_name ( index , <EOL> include_schema = False ) , <EOL> preparer . format_table ( index . table ) <EOL> ) <EOL> using = index . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if using : <EOL> text += "<STR_LIT>" % preparer . quote ( using ) <EOL> ops = index . dialect_options [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> text += "<STR_LIT>" % ( <EOL> '<STR_LIT:U+002CU+0020>' . join ( [ <EOL> self . sql_compiler . process ( <EOL> expr . self_group ( ) <EOL> if not isinstance ( expr , expression . ColumnClause ) <EOL> else expr , <EOL> include_table = False , literal_binds = True ) + <EOL> ( <EOL> ( '<STR_LIT:U+0020>' + ops [ expr . key ] ) <EOL> if hasattr ( expr , '<STR_LIT:key>' ) <EOL> and expr . key in ops else '<STR_LIT>' <EOL> ) <EOL> for expr in index . expressions <EOL> ] ) <EOL> ) <EOL> whereclause = index . dialect_options [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> if whereclause is not None : <EOL> where_compiled = self . sql_compiler . process ( <EOL> whereclause , include_table = False , <EOL> literal_binds = True ) <EOL> text += "<STR_LIT>" + where_compiled <EOL> return text <EOL> def visit_exclude_constraint ( self , constraint ) : <EOL> text = "<STR_LIT>" <EOL> if constraint . name is not None : <EOL> text += "<STR_LIT>" % self . preparer . format_constraint ( constraint ) <EOL> elements = [ ] <EOL> for c in constraint . columns : <EOL> op = constraint . operators [ c . name ] <EOL> elements . append ( self . preparer . quote ( c . name ) + '<STR_LIT>' + op ) <EOL> text += "<STR_LIT>" % ( constraint . using , <EOL> '<STR_LIT:U+002CU+0020>' . join ( elements ) ) <EOL> if constraint . where is not None : <EOL> text += '<STR_LIT>' % self . sql_compiler . process ( <EOL> constraint . where , <EOL> literal_binds = True ) <EOL> text += self . define_constraint_deferrability ( constraint ) <EOL> return text <EOL> class PGTypeCompiler ( compiler . GenericTypeCompiler ) : <EOL> def visit_TSVECTOR ( self , type ) : <EOL> return "<STR_LIT>" <EOL> def visit_INET ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_CIDR ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_MACADDR ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_OID ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_FLOAT ( self , type_ ) : <EOL> if not type_ . precision : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" % { '<STR_LIT>' : type_ . precision } <EOL> def visit_DOUBLE_PRECISION ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_BIGINT ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_HSTORE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_JSON ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_JSONB ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_INT4RANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_INT8RANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_NUMRANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_DATERANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_TSRANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_TSTZRANGE ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_datetime ( self , type_ ) : <EOL> return self . visit_TIMESTAMP ( type_ ) <EOL> def visit_enum ( self , type_ ) : <EOL> if not type_ . native_enum or not self . dialect . supports_native_enum : <EOL> return super ( PGTypeCompiler , self ) . visit_enum ( type_ ) <EOL> else : <EOL> return self . visit_ENUM ( type_ ) <EOL> def visit_ENUM ( self , type_ ) : <EOL> return self . dialect . identifier_preparer . format_type ( type_ ) <EOL> def visit_TIMESTAMP ( self , type_ ) : <EOL> return "<STR_LIT>" % ( <EOL> getattr ( type_ , '<STR_LIT>' , None ) and "<STR_LIT>" % <EOL> type_ . precision or "<STR_LIT>" , <EOL> ( type_ . timezone and "<STR_LIT>" or "<STR_LIT>" ) + "<STR_LIT>" <EOL> ) <EOL> def visit_TIME ( self , type_ ) : <EOL> return "<STR_LIT>" % ( <EOL> getattr ( type_ , '<STR_LIT>' , None ) and "<STR_LIT>" % <EOL> type_ . precision or "<STR_LIT>" , <EOL> ( type_ . timezone and "<STR_LIT>" or "<STR_LIT>" ) + "<STR_LIT>" <EOL> ) <EOL> def visit_INTERVAL ( self , type_ ) : <EOL> if type_ . precision is not None : <EOL> return "<STR_LIT>" % type_ . precision <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_BIT ( self , type_ ) : <EOL> if type_ . varying : <EOL> compiled = "<STR_LIT>" <EOL> if type_ . length is not None : <EOL> compiled += "<STR_LIT>" % type_ . length <EOL> else : <EOL> compiled = "<STR_LIT>" % type_ . length <EOL> return compiled <EOL> def visit_UUID ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_large_binary ( self , type_ ) : <EOL> return self . visit_BYTEA ( type_ ) <EOL> def visit_BYTEA ( self , type_ ) : <EOL> return "<STR_LIT>" <EOL> def visit_ARRAY ( self , type_ ) : <EOL> return self . process ( type_ . item_type ) + ( '<STR_LIT>' * ( type_ . dimensions <EOL> if type_ . dimensions <EOL> is not None else <NUM_LIT:1> ) ) <EOL> class PGIdentifierPreparer ( compiler . IdentifierPreparer ) : <EOL> reserved_words = RESERVED_WORDS <EOL> def _unquote_identifier ( self , value ) : <EOL> if value [ <NUM_LIT:0> ] == self . initial_quote : <EOL> value = value [ <NUM_LIT:1> : - <NUM_LIT:1> ] . replace ( self . escape_to_quote , self . escape_quote ) <EOL> return value <EOL> def format_type ( self , type_ , use_schema = True ) : <EOL> if not type_ . name : <EOL> raise exc . CompileError ( "<STR_LIT>" ) <EOL> name = self . quote ( type_ . name ) <EOL> if not self . omit_schema and use_schema and type_ . schema is not None : <EOL> name = self . quote_schema ( type_ . schema ) + "<STR_LIT:.>" + name <EOL> return name <EOL> class PGInspector ( reflection . Inspector ) : <EOL> def __init__ ( self , conn ) : <EOL> reflection . Inspector . __init__ ( self , conn ) <EOL> def get_table_oid ( self , table_name , schema = None ) : <EOL> """<STR_LIT>""" <EOL> return self . dialect . get_table_oid ( self . bind , table_name , schema , <EOL> info_cache = self . info_cache ) <EOL> class CreateEnumType ( schema . _CreateDropBase ) : <EOL> __visit_name__ = "<STR_LIT>" <EOL> class DropEnumType ( schema . _CreateDropBase ) : <EOL> __visit_name__ = "<STR_LIT>" <EOL> class PGExecutionContext ( default . DefaultExecutionContext ) : <EOL> def fire_sequence ( self , seq , type_ ) : <EOL> return self . _execute_scalar ( ( <EOL> "<STR_LIT>" % <EOL> self . dialect . identifier_preparer . format_sequence ( seq ) ) , type_ ) <EOL> def get_insert_default ( self , column ) : <EOL> if column . primary_key and column is column . table . _autoincrement_column : <EOL> if column . server_default and column . server_default . has_argument : <EOL> return self . _execute_scalar ( "<STR_LIT>" % <EOL> column . server_default . arg , <EOL> column . type ) <EOL> elif ( column . default is None or <EOL> ( column . default . is_sequence and <EOL> column . default . optional ) ) : <EOL> try : <EOL> seq_name = column . _postgresql_seq_name <EOL> except AttributeError : <EOL> tab = column . table . name <EOL> col = column . name <EOL> tab = tab [ <NUM_LIT:0> : <NUM_LIT> + max ( <NUM_LIT:0> , ( <NUM_LIT> - len ( col ) ) ) ] <EOL> col = col [ <NUM_LIT:0> : <NUM_LIT> + max ( <NUM_LIT:0> , ( <NUM_LIT> - len ( tab ) ) ) ] <EOL> name = "<STR_LIT>" % ( tab , col ) <EOL> column . _postgresql_seq_name = seq_name = name <EOL> sch = column . table . schema <EOL> if sch is not None : <EOL> exc = "<STR_LIT>" % ( sch , seq_name ) <EOL> else : <EOL> exc = "<STR_LIT>" % ( seq_name , ) <EOL> return self . _execute_scalar ( exc , column . type ) <EOL> return super ( PGExecutionContext , self ) . get_insert_default ( column ) <EOL> class PGDialect ( default . DefaultDialect ) : <EOL> name = '<STR_LIT>' <EOL> supports_alter = True <EOL> max_identifier_length = <NUM_LIT> <EOL> supports_sane_rowcount = True <EOL> supports_native_enum = True <EOL> supports_native_boolean = True <EOL> supports_smallserial = True <EOL> supports_sequences = True <EOL> sequences_optional = True <EOL> preexecute_autoincrement_sequences = True <EOL> postfetch_lastrowid = False <EOL> supports_default_values = True <EOL> supports_empty_insert = False <EOL> supports_multivalues_insert = True <EOL> default_paramstyle = '<STR_LIT>' <EOL> ischema_names = ischema_names <EOL> colspecs = colspecs <EOL> statement_compiler = PGCompiler <EOL> ddl_compiler = PGDDLCompiler <EOL> type_compiler = PGTypeCompiler <EOL> preparer = PGIdentifierPreparer <EOL> execution_ctx_cls = PGExecutionContext <EOL> inspector = PGInspector <EOL> isolation_level = None <EOL> construct_arguments = [ <EOL> ( schema . Index , { <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : False , <EOL> } ) , <EOL> ( schema . Table , { <EOL> "<STR_LIT>" : False <EOL> } ) <EOL> ] <EOL> reflection_options = ( '<STR_LIT>' , ) <EOL> _backslash_escapes = True <EOL> def __init__ ( self , isolation_level = None , json_serializer = None , <EOL> json_deserializer = None , ** kwargs ) : <EOL> default . DefaultDialect . __init__ ( self , ** kwargs ) <EOL> self . isolation_level = isolation_level <EOL> self . _json_deserializer = json_deserializer <EOL> self . _json_serializer = json_serializer <EOL> def initialize ( self , connection ) : <EOL> super ( PGDialect , self ) . initialize ( connection ) <EOL> self . implicit_returning = self . server_version_info > ( <NUM_LIT:8> , <NUM_LIT:2> ) and self . __dict__ . get ( '<STR_LIT>' , True ) <EOL> self . supports_native_enum = self . server_version_info >= ( <NUM_LIT:8> , <NUM_LIT:3> ) <EOL> if not self . supports_native_enum : <EOL> self . colspecs = self . colspecs . copy ( ) <EOL> self . colspecs . pop ( sqltypes . Enum , None ) <EOL> self . colspecs . pop ( ENUM , None ) <EOL> self . supports_smallserial = self . server_version_info >= ( <NUM_LIT:9> , <NUM_LIT:2> ) <EOL> self . _backslash_escapes = self . server_version_info < ( <NUM_LIT:8> , <NUM_LIT:2> ) or connection . scalar ( <EOL> "<STR_LIT>" <EOL> ) == '<STR_LIT>' <EOL> def on_connect ( self ) : <EOL> if self . isolation_level is not None : <EOL> def connect ( conn ) : <EOL> self . set_isolation_level ( conn , self . isolation_level ) <EOL> return connect <EOL> else : <EOL> return None <EOL> _isolation_lookup = set ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def set_isolation_level ( self , connection , level ) : <EOL> level = level . replace ( '<STR_LIT:_>' , '<STR_LIT:U+0020>' ) <EOL> if level not in self . _isolation_lookup : <EOL> raise exc . ArgumentError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( level , self . name , "<STR_LIT:U+002CU+0020>" . join ( self . _isolation_lookup ) ) <EOL> ) <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % level ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> cursor . close ( ) <EOL> def get_isolation_level ( self , connection ) : <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> val = cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> cursor . close ( ) <EOL> return val . upper ( ) <EOL> def do_begin_twophase ( self , connection , xid ) : <EOL> self . do_begin ( connection . connection ) <EOL> def do_prepare_twophase ( self , connection , xid ) : <EOL> connection . execute ( "<STR_LIT>" % xid ) <EOL> def do_rollback_twophase ( self , connection , xid , <EOL> is_prepared = True , recover = False ) : <EOL> if is_prepared : <EOL> if recover : <EOL> connection . execute ( "<STR_LIT>" ) <EOL> connection . execute ( "<STR_LIT>" % xid ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> self . do_rollback ( connection . connection ) <EOL> else : <EOL> self . do_rollback ( connection . connection ) <EOL> def do_commit_twophase ( self , connection , xid , <EOL> is_prepared = True , recover = False ) : <EOL> if is_prepared : <EOL> if recover : <EOL> connection . execute ( "<STR_LIT>" ) <EOL> connection . execute ( "<STR_LIT>" % xid ) <EOL> connection . execute ( "<STR_LIT>" ) <EOL> self . do_rollback ( connection . connection ) <EOL> else : <EOL> self . do_commit ( connection . connection ) <EOL> def do_recover_twophase ( self , connection ) : <EOL> resultset = connection . execute ( <EOL> sql . text ( "<STR_LIT>" ) ) <EOL> return [ row [ <NUM_LIT:0> ] for row in resultset ] <EOL> def _get_default_schema_name ( self , connection ) : <EOL> return connection . scalar ( "<STR_LIT>" ) <EOL> def has_schema ( self , connection , schema ) : <EOL> query = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> cursor = connection . execute ( <EOL> sql . text ( <EOL> query , <EOL> bindparams = [ <EOL> sql . bindparam ( <EOL> '<STR_LIT>' , util . text_type ( schema . lower ( ) ) , <EOL> type_ = sqltypes . Unicode ) ] <EOL> ) <EOL> ) <EOL> return bool ( cursor . first ( ) ) <EOL> def has_table ( self , connection , table_name , schema = None ) : <EOL> if schema is None : <EOL> cursor = connection . execute ( <EOL> sql . text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bindparams = [ <EOL> sql . bindparam ( '<STR_LIT:name>' , util . text_type ( table_name ) , <EOL> type_ = sqltypes . Unicode ) ] <EOL> ) <EOL> ) <EOL> else : <EOL> cursor = connection . execute ( <EOL> sql . text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bindparams = [ <EOL> sql . bindparam ( '<STR_LIT:name>' , <EOL> util . text_type ( table_name ) , <EOL> type_ = sqltypes . Unicode ) , <EOL> sql . bindparam ( '<STR_LIT>' , <EOL> util . text_type ( schema ) , <EOL> type_ = sqltypes . Unicode ) ] <EOL> ) <EOL> ) <EOL> return bool ( cursor . first ( ) ) <EOL> def has_sequence ( self , connection , sequence_name , schema = None ) : <EOL> if schema is None : <EOL> cursor = connection . execute ( <EOL> sql . text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bindparams = [ <EOL> sql . bindparam ( '<STR_LIT:name>' , util . text_type ( sequence_name ) , <EOL> type_ = sqltypes . Unicode ) <EOL> ] <EOL> ) <EOL> ) <EOL> else : <EOL> cursor = connection . execute ( <EOL> sql . text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bindparams = [ <EOL> sql . bindparam ( '<STR_LIT:name>' , util . text_type ( sequence_name ) , <EOL> type_ = sqltypes . Unicode ) , <EOL> sql . bindparam ( '<STR_LIT>' , <EOL> util . text_type ( schema ) , <EOL> type_ = sqltypes . Unicode ) <EOL> ] <EOL> ) <EOL> ) <EOL> return bool ( cursor . first ( ) ) <EOL> def has_type ( self , connection , type_name , schema = None ) : <EOL> if schema is not None : <EOL> query = """<STR_LIT>""" <EOL> query = sql . text ( query ) <EOL> else : <EOL> query = """<STR_LIT>""" <EOL> query = sql . text ( query ) <EOL> query = query . bindparams ( <EOL> sql . bindparam ( '<STR_LIT>' , <EOL> util . text_type ( type_name ) , type_ = sqltypes . Unicode ) , <EOL> ) <EOL> if schema is not None : <EOL> query = query . bindparams ( <EOL> sql . bindparam ( '<STR_LIT>' , <EOL> util . text_type ( schema ) , type_ = sqltypes . Unicode ) , <EOL> ) <EOL> cursor = connection . execute ( query ) <EOL> return bool ( cursor . scalar ( ) ) <EOL> def _get_server_version_info ( self , connection ) : <EOL> v = connection . execute ( "<STR_LIT>" ) . scalar ( ) <EOL> m = re . match ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> v ) <EOL> if not m : <EOL> raise AssertionError ( <EOL> "<STR_LIT>" % v ) <EOL> return tuple ( [ int ( x ) for x in m . group ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) if x is not None ] ) <EOL> @ reflection . cache <EOL> def get_table_oid ( self , connection , table_name , schema = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> table_oid = None <EOL> if schema is not None : <EOL> schema_where_clause = "<STR_LIT>" <EOL> else : <EOL> schema_where_clause = "<STR_LIT>" <EOL> query = """<STR_LIT>""" % schema_where_clause <EOL> table_name = util . text_type ( table_name ) <EOL> if schema is not None : <EOL> schema = util . text_type ( schema ) <EOL> s = sql . text ( query ) . bindparams ( table_name = sqltypes . Unicode ) <EOL> s = s . columns ( oid = sqltypes . Integer ) <EOL> if schema : <EOL> s = s . bindparams ( sql . bindparam ( '<STR_LIT>' , type_ = sqltypes . Unicode ) ) <EOL> c = connection . execute ( s , table_name = table_name , schema = schema ) <EOL> table_oid = c . scalar ( ) <EOL> if table_oid is None : <EOL> raise exc . NoSuchTableError ( table_name ) <EOL> return table_oid <EOL> @ reflection . cache <EOL> def get_schema_names ( self , connection , ** kw ) : <EOL> s = """<STR_LIT>""" <EOL> rp = connection . execute ( s ) <EOL> if util . py2k : <EOL> schema_names = [ row [ <NUM_LIT:0> ] . decode ( self . encoding ) for row in rp <EOL> if not row [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) ] <EOL> else : <EOL> schema_names = [ row [ <NUM_LIT:0> ] for row in rp <EOL> if not row [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) ] <EOL> return schema_names <EOL> @ reflection . cache <EOL> def get_table_names ( self , connection , schema = None , ** kw ) : <EOL> if schema is not None : <EOL> current_schema = schema <EOL> else : <EOL> current_schema = self . default_schema_name <EOL> result = connection . execute ( <EOL> sql . text ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> current_schema , <EOL> typemap = { '<STR_LIT>' : sqltypes . Unicode } <EOL> ) <EOL> ) <EOL> return [ row [ <NUM_LIT:0> ] for row in result ] <EOL> @ reflection . cache <EOL> def get_view_names ( self , connection , schema = None , ** kw ) : <EOL> if schema is not None : <EOL> current_schema = schema <EOL> else : <EOL> current_schema = self . default_schema_name <EOL> s = """<STR_LIT>""" % dict ( schema = current_schema ) <EOL> if util . py2k : <EOL> view_names = [ row [ <NUM_LIT:0> ] . decode ( self . encoding ) <EOL> for row in connection . execute ( s ) ] <EOL> else : <EOL> view_names = [ row [ <NUM_LIT:0> ] for row in connection . execute ( s ) ] <EOL> return view_names <EOL> @ reflection . cache <EOL> def get_view_definition ( self , connection , view_name , schema = None , ** kw ) : <EOL> if schema is not None : <EOL> current_schema = schema <EOL> else : <EOL> current_schema = self . default_schema_name <EOL> s = """<STR_LIT>""" <EOL> rp = connection . execute ( sql . text ( s ) , <EOL> view_name = view_name , schema = current_schema ) <EOL> if rp : <EOL> if util . py2k : <EOL> view_def = rp . scalar ( ) . decode ( self . encoding ) <EOL> else : <EOL> view_def = rp . scalar ( ) <EOL> return view_def <EOL> @ reflection . cache <EOL> def get_columns ( self , connection , table_name , schema = None , ** kw ) : <EOL> table_oid = self . get_table_oid ( connection , table_name , schema , <EOL> info_cache = kw . get ( '<STR_LIT>' ) ) <EOL> SQL_COLS = """<STR_LIT>""" <EOL> s = sql . text ( SQL_COLS , <EOL> bindparams = [ <EOL> sql . bindparam ( '<STR_LIT>' , type_ = sqltypes . Integer ) ] , <EOL> typemap = { <EOL> '<STR_LIT>' : sqltypes . Unicode , <EOL> '<STR_LIT:default>' : sqltypes . Unicode } <EOL> ) <EOL> c = connection . execute ( s , table_oid = table_oid ) <EOL> rows = c . fetchall ( ) <EOL> domains = self . _load_domains ( connection ) <EOL> enums = self . _load_enums ( connection ) <EOL> columns = [ ] <EOL> for name , format_type , default , notnull , attnum , table_oid in rows : <EOL> column_info = self . _get_column_info ( <EOL> name , format_type , default , notnull , domains , enums , schema ) <EOL> columns . append ( column_info ) <EOL> return columns <EOL> def _get_column_info ( self , name , format_type , default , <EOL> notnull , domains , enums , schema ) : <EOL> attype = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , format_type ) <EOL> attype = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , attype ) <EOL> nullable = not notnull <EOL> is_array = format_type . endswith ( '<STR_LIT>' ) <EOL> charlen = re . search ( '<STR_LIT>' , format_type ) <EOL> if charlen : <EOL> charlen = charlen . group ( <NUM_LIT:1> ) <EOL> args = re . search ( '<STR_LIT>' , format_type ) <EOL> if args and args . group ( <NUM_LIT:1> ) : <EOL> args = tuple ( re . split ( '<STR_LIT>' , args . group ( <NUM_LIT:1> ) ) ) <EOL> else : <EOL> args = ( ) <EOL> kwargs = { } <EOL> if attype == '<STR_LIT>' : <EOL> if charlen : <EOL> prec , scale = charlen . split ( '<STR_LIT:U+002C>' ) <EOL> args = ( int ( prec ) , int ( scale ) ) <EOL> else : <EOL> args = ( ) <EOL> elif attype == '<STR_LIT>' : <EOL> args = ( <NUM_LIT> , ) <EOL> elif attype == '<STR_LIT>' : <EOL> args = ( ) <EOL> elif attype in ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> if charlen : <EOL> kwargs [ '<STR_LIT>' ] = int ( charlen ) <EOL> args = ( ) <EOL> elif attype in ( '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:time>' ) : <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> if charlen : <EOL> kwargs [ '<STR_LIT>' ] = int ( charlen ) <EOL> args = ( ) <EOL> elif attype == '<STR_LIT>' : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> if charlen : <EOL> args = ( int ( charlen ) , ) <EOL> else : <EOL> args = ( ) <EOL> elif attype in ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> if charlen : <EOL> kwargs [ '<STR_LIT>' ] = int ( charlen ) <EOL> args = ( ) <EOL> elif charlen : <EOL> args = ( int ( charlen ) , ) <EOL> while True : <EOL> if attype in self . ischema_names : <EOL> coltype = self . ischema_names [ attype ] <EOL> break <EOL> elif attype in enums : <EOL> enum = enums [ attype ] <EOL> coltype = ENUM <EOL> if "<STR_LIT:.>" in attype : <EOL> kwargs [ '<STR_LIT>' ] , kwargs [ '<STR_LIT:name>' ] = attype . split ( '<STR_LIT:.>' ) <EOL> else : <EOL> kwargs [ '<STR_LIT:name>' ] = attype <EOL> args = tuple ( enum [ '<STR_LIT>' ] ) <EOL> break <EOL> elif attype in domains : <EOL> domain = domains [ attype ] <EOL> attype = domain [ '<STR_LIT>' ] <EOL> nullable = domain [ '<STR_LIT>' ] <EOL> if domain [ '<STR_LIT:default>' ] and not default : <EOL> default = domain [ '<STR_LIT:default>' ] <EOL> continue <EOL> else : <EOL> coltype = None <EOL> break <EOL> if coltype : <EOL> coltype = coltype ( * args , ** kwargs ) <EOL> if is_array : <EOL> coltype = ARRAY ( coltype ) <EOL> else : <EOL> util . warn ( "<STR_LIT>" % <EOL> ( attype , name ) ) <EOL> coltype = sqltypes . NULLTYPE <EOL> autoincrement = False <EOL> if default is not None : <EOL> match = re . search ( r"""<STR_LIT>""" , default ) <EOL> if match is not None : <EOL> autoincrement = True <EOL> sch = schema <EOL> if '<STR_LIT:.>' not in match . group ( <NUM_LIT:2> ) and sch is not None : <EOL> default = match . group ( <NUM_LIT:1> ) + ( '<STR_LIT>' % sch ) + '<STR_LIT:.>' + match . group ( <NUM_LIT:2> ) + match . group ( <NUM_LIT:3> ) <EOL> column_info = dict ( name = name , type = coltype , nullable = nullable , <EOL> default = default , autoincrement = autoincrement ) <EOL> return column_info <EOL> @ reflection . cache <EOL> def get_pk_constraint ( self , connection , table_name , schema = None , ** kw ) : <EOL> table_oid = self . get_table_oid ( connection , table_name , schema , <EOL> info_cache = kw . get ( '<STR_LIT>' ) ) <EOL> if self . server_version_info < ( <NUM_LIT:8> , <NUM_LIT:4> ) : <EOL> PK_SQL = """<STR_LIT>""" % self . _pg_index_any ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> else : <EOL> PK_SQL = """<STR_LIT>""" <EOL> t = sql . text ( PK_SQL , typemap = { '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( t , table_oid = table_oid ) <EOL> cols = [ r [ <NUM_LIT:0> ] for r in c . fetchall ( ) ] <EOL> PK_CONS_SQL = """<STR_LIT>""" <EOL> t = sql . text ( PK_CONS_SQL , typemap = { '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( t , table_oid = table_oid ) <EOL> name = c . scalar ( ) <EOL> return { '<STR_LIT>' : cols , '<STR_LIT:name>' : name } <EOL> @ reflection . cache <EOL> def get_foreign_keys ( self , connection , table_name , schema = None , <EOL> postgresql_ignore_search_path = False , ** kw ) : <EOL> preparer = self . identifier_preparer <EOL> table_oid = self . get_table_oid ( connection , table_name , schema , <EOL> info_cache = kw . get ( '<STR_LIT>' ) ) <EOL> FK_SQL = """<STR_LIT>""" <EOL> FK_REGEX = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> ) <EOL> t = sql . text ( FK_SQL , typemap = { <EOL> '<STR_LIT>' : sqltypes . Unicode , <EOL> '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( t , table = table_oid ) <EOL> fkeys = [ ] <EOL> for conname , condef , conschema in c . fetchall ( ) : <EOL> m = re . search ( FK_REGEX , condef ) . groups ( ) <EOL> constrained_columns , referred_schema , referred_table , referred_columns , _ , match , _ , onupdate , _ , ondelete , deferrable , _ , initially = m <EOL> if deferrable is not None : <EOL> deferrable = True if deferrable == '<STR_LIT>' else False <EOL> constrained_columns = [ preparer . _unquote_identifier ( x ) <EOL> for x in re . split ( <EOL> r'<STR_LIT>' , constrained_columns ) ] <EOL> if postgresql_ignore_search_path : <EOL> if conschema != self . default_schema_name : <EOL> referred_schema = conschema <EOL> else : <EOL> referred_schema = schema <EOL> elif referred_schema : <EOL> referred_schema = preparer . _unquote_identifier ( referred_schema ) <EOL> elif schema is not None and schema == conschema : <EOL> referred_schema = schema <EOL> referred_table = preparer . _unquote_identifier ( referred_table ) <EOL> referred_columns = [ preparer . _unquote_identifier ( x ) <EOL> for x in <EOL> re . split ( r'<STR_LIT>' , referred_columns ) ] <EOL> fkey_d = { <EOL> '<STR_LIT:name>' : conname , <EOL> '<STR_LIT>' : constrained_columns , <EOL> '<STR_LIT>' : referred_schema , <EOL> '<STR_LIT>' : referred_table , <EOL> '<STR_LIT>' : referred_columns , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : onupdate , <EOL> '<STR_LIT>' : ondelete , <EOL> '<STR_LIT>' : deferrable , <EOL> '<STR_LIT>' : initially , <EOL> '<STR_LIT>' : match <EOL> } <EOL> } <EOL> fkeys . append ( fkey_d ) <EOL> return fkeys <EOL> def _pg_index_any ( self , col , compare_to ) : <EOL> if self . server_version_info < ( <NUM_LIT:8> , <NUM_LIT:1> ) : <EOL> return "<STR_LIT>" % "<STR_LIT>" . join ( <EOL> "<STR_LIT>" % ( compare_to , ind , col ) <EOL> for ind in range ( <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> ) <EOL> else : <EOL> return "<STR_LIT>" % ( col , compare_to ) <EOL> @ reflection . cache <EOL> def get_indexes ( self , connection , table_name , schema , ** kw ) : <EOL> table_oid = self . get_table_oid ( connection , table_name , schema , <EOL> info_cache = kw . get ( '<STR_LIT>' ) ) <EOL> IDX_SQL = """<STR_LIT>""" % ( <EOL> "<STR_LIT>" if self . server_version_info >= ( <NUM_LIT:8> , <NUM_LIT:3> ) else "<STR_LIT>" , <EOL> self . _pg_index_any ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> ) <EOL> t = sql . text ( IDX_SQL , typemap = { '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( t , table_oid = table_oid ) <EOL> indexes = defaultdict ( lambda : defaultdict ( dict ) ) <EOL> sv_idx_name = None <EOL> for row in c . fetchall ( ) : <EOL> idx_name , unique , expr , prd , col , col_num , idx_key = row <EOL> if expr : <EOL> if idx_name != sv_idx_name : <EOL> util . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % idx_name ) <EOL> sv_idx_name = idx_name <EOL> continue <EOL> if prd and not idx_name == sv_idx_name : <EOL> util . warn ( <EOL> "<STR_LIT>" <EOL> % idx_name ) <EOL> sv_idx_name = idx_name <EOL> index = indexes [ idx_name ] <EOL> if col is not None : <EOL> index [ '<STR_LIT>' ] [ col_num ] = col <EOL> index [ '<STR_LIT:key>' ] = [ int ( k . strip ( ) ) for k in idx_key . split ( ) ] <EOL> index [ '<STR_LIT>' ] = unique <EOL> return [ <EOL> { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : idx [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ idx [ '<STR_LIT>' ] [ i ] for i in idx [ '<STR_LIT:key>' ] ] } <EOL> for name , idx in indexes . items ( ) <EOL> ] <EOL> @ reflection . cache <EOL> def get_unique_constraints ( self , connection , table_name , <EOL> schema = None , ** kw ) : <EOL> table_oid = self . get_table_oid ( connection , table_name , schema , <EOL> info_cache = kw . get ( '<STR_LIT>' ) ) <EOL> UNIQUE_SQL = """<STR_LIT>""" <EOL> t = sql . text ( UNIQUE_SQL , typemap = { '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( t , table_oid = table_oid ) <EOL> uniques = defaultdict ( lambda : defaultdict ( dict ) ) <EOL> for row in c . fetchall ( ) : <EOL> uc = uniques [ row . name ] <EOL> uc [ "<STR_LIT:key>" ] = row . key <EOL> uc [ "<STR_LIT>" ] [ row . col_num ] = row . col_name <EOL> return [ <EOL> { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : [ uc [ "<STR_LIT>" ] [ i ] for i in uc [ "<STR_LIT:key>" ] ] } <EOL> for name , uc in uniques . items ( ) <EOL> ] <EOL> def _load_enums ( self , connection ) : <EOL> if not self . supports_native_enum : <EOL> return { } <EOL> SQL_ENUMS = """<STR_LIT>""" <EOL> s = sql . text ( SQL_ENUMS , typemap = { <EOL> '<STR_LIT>' : sqltypes . Unicode , <EOL> '<STR_LIT:label>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( s ) <EOL> enums = { } <EOL> for enum in c . fetchall ( ) : <EOL> if enum [ '<STR_LIT>' ] : <EOL> name = enum [ '<STR_LIT:name>' ] <EOL> else : <EOL> name = "<STR_LIT>" % ( enum [ '<STR_LIT>' ] , enum [ '<STR_LIT:name>' ] ) <EOL> if name in enums : <EOL> enums [ name ] [ '<STR_LIT>' ] . append ( enum [ '<STR_LIT:label>' ] ) <EOL> else : <EOL> enums [ name ] = { <EOL> '<STR_LIT>' : [ enum [ '<STR_LIT:label>' ] ] , <EOL> } <EOL> return enums <EOL> def _load_domains ( self , connection ) : <EOL> SQL_DOMAINS = """<STR_LIT>""" <EOL> s = sql . text ( SQL_DOMAINS , typemap = { '<STR_LIT>' : sqltypes . Unicode } ) <EOL> c = connection . execute ( s ) <EOL> domains = { } <EOL> for domain in c . fetchall ( ) : <EOL> attype = re . search ( '<STR_LIT>' , domain [ '<STR_LIT>' ] ) . group ( <NUM_LIT:1> ) <EOL> if domain [ '<STR_LIT>' ] : <EOL> name = domain [ '<STR_LIT:name>' ] <EOL> else : <EOL> name = "<STR_LIT>" % ( domain [ '<STR_LIT>' ] , domain [ '<STR_LIT:name>' ] ) <EOL> domains [ name ] = { <EOL> '<STR_LIT>' : attype , <EOL> '<STR_LIT>' : domain [ '<STR_LIT>' ] , <EOL> '<STR_LIT:default>' : domain [ '<STR_LIT:default>' ] <EOL> } <EOL> return domains </s>
