我们从Python开源项目中，提取了以下28个代码示例，用于说明如何使用sqlalchemy.ext.compiler.compiles()。
def compiles(class_, *specs): """Register a function as a compiler for a given :class:`.ClauseElement` type.""" def decorate(fn): existing = class_.__dict__.get('_compiler_dispatcher', None) existing_dispatch = class_.__dict__.get('_compiler_dispatch') if not existing: existing = _dispatcher() if existing_dispatch: existing.specs['default'] = existing_dispatch # TODO: why is the lambda needed ? setattr(class_, '_compiler_dispatch', lambda *arg, **kw: existing(*arg, **kw)) setattr(class_, '_compiler_dispatcher', existing) if specs: for s in specs: existing.specs[s] = fn else: existing.specs['default'] = fn return fn return decorate
def compiles(class_, *specs): """Register a function as a compiler for a given :class:`.ClauseElement` type.""" def decorate(fn): # get an existing @compiles handler existing = class_.__dict__.get('_compiler_dispatcher', None) # get the original handler. All ClauseElement classes have one # of these, but some TypeEngine classes will not. existing_dispatch = getattr(class_, '_compiler_dispatch', None) if not existing: existing = _dispatcher() if existing_dispatch: def _wrap_existing_dispatch(element, compiler, **kw): try: return existing_dispatch(element, compiler, **kw) except exc.UnsupportedCompilationError: raise exc.CompileError( "%s construct has no default " "compilation handler." % type(element)) existing.specs['default'] = _wrap_existing_dispatch # TODO: why is the lambda needed ? setattr(class_, '_compiler_dispatch', lambda *arg, **kw: existing(*arg, **kw)) setattr(class_, '_compiler_dispatcher', existing) if specs: for s in specs: existing.specs[s] = fn else: existing.specs['default'] = fn return fn return decorate
def _is_sqlalchemy_connectable(con): global _SQLALCHEMY_INSTALLED if _SQLALCHEMY_INSTALLED is None: try: import sqlalchemy _SQLALCHEMY_INSTALLED = True from distutils.version import LooseVersion ver = LooseVersion(sqlalchemy.__version__) # For sqlalchemy versions < 0.8.2, the BIGINT type is recognized # for a sqlite engine, which results in a warning when trying to # read/write a DataFrame with int64 values. (GH7433) if ver < '0.8.2': from sqlalchemy import BigInteger from sqlalchemy.ext.compiler import compiles @compiles(BigInteger, 'sqlite') def compile_big_int_sqlite(type_, compiler, **kw): return 'INTEGER' except ImportError: _SQLALCHEMY_INSTALLED = False if _SQLALCHEMY_INSTALLED: import sqlalchemy return isinstance(con, sqlalchemy.engine.Connectable) else: return False
def visit_BYTEINT(self, type_, **kw): return 'BYTEINT' #@compiles(Select, 'teradata') #def compile_select(element, compiler, **kw): # """ # """ # # if not getattr(element, '_window_visit', None): # if element._limit is not None or element._offset is not None: # limit, offset = element._limit, element._offset # # orderby=compiler.process(element._order_by_clause) # if orderby: # element = element._generate() # element._window_visit=True # #element._limit = None # #element._offset = None cant set to none... # # # add a ROW NUMBER() OVER(ORDER BY) column # element = element.column(sql.literal_column('ROW NUMBER() OVER (ORDER BY %s)' % orderby).label('rownum')).order_by(None) # # # wrap into a subquery # limitselect = sql.select([c for c in element.alias().c if c.key != 'rownum']) # # limitselect._window_visit=True # limitselect._is_wrapper=True # # if offset is not None: # limitselect.append_whereclause(sql.column('rownum') > offset) # if limit is not None: # limitselect.append_whereclause(sql.column('rownum') <= (limit + offset)) # else: # limitselect.append_whereclause(sql.column("rownum") <= limit) # # element = limitselect # # kw['iswrapper'] = getattr(element, '_is_wrapper', False) # return compiler.visit_select(element, **kw)
