identifiers, such as 'database.owner'. [ticket: 594, 1341]
- sql
+ - Back-ported the "compiler" extension from SQLA 0.6. This
+ is a standardized interface which allows the creation of custom
+ ClauseElement subclasses and compilers. In particular it's
+ handy as an alternative to text() when you'd like to
+ build a construct that has database-specific compilations.
+ See the extension docs for details.
+
- Exception messages are truncated when the list of bound
parameters is larger than 10, preventing enormous
multi-page exceptions from filling up screens and logfiles
--- /dev/null
+compiler
+========
+
+.. automodule:: sqlalchemy.ext.compiler
+ :members:
\ No newline at end of file
orderinglist
serializer
sqlsoup
+ compiler
class DefaultDialect(base.Dialect):
"""Default implementation of Dialect"""
+ name = 'default'
schemagenerator = compiler.SchemaGenerator
schemadropper = compiler.SchemaDropper
statement_compiler = compiler.DefaultCompiler
--- /dev/null
+"""Provides an API for creation of custom ClauseElements and compilers.
+
+Synopsis
+========
+
+Usage involves the creation of one or more :class:`~sqlalchemy.sql.expression.ClauseElement`
+subclasses and one or more callables defining its compilation::
+
+ from sqlalchemy.ext.compiler import compiles
+ from sqlalchemy.sql.expression import ColumnClause
+
+ class MyColumn(ColumnClause):
+ pass
+
+ @compiles(MyColumn)
+ def compile_mycolumn(element, compiler, **kw):
+ return "[%s]" % element.name
+
+Above, ``MyColumn`` extends :class:`~sqlalchemy.sql.expression.ColumnClause`, the
+base expression element for column objects. The ``compiles`` decorator registers
+itself with the ``MyColumn`` class so that it is invoked when the object
+is compiled to a string::
+
+ from sqlalchemy import select
+
+ s = select([MyColumn('x'), MyColumn('y')])
+ print str(s)
+
+Produces::
+
+ SELECT [x], [y]
+
+Compilers can also be made dialect-specific. The appropriate compiler will be invoked
+for the dialect in use::
+
+ from sqlalchemy.schema import DDLElement # this is a SQLA 0.6 construct
+
+ class AlterColumn(DDLElement):
+
+ def __init__(self, column, cmd):
+ self.column = column
+ self.cmd = cmd
+
+ @compiles(AlterColumn)
+ def visit_alter_column(element, compiler, **kw):
+ return "ALTER COLUMN %s ..." % element.column.name
+
+ @compiles(AlterColumn, 'postgres')
+ def visit_alter_column(element, compiler, **kw):
+ return "ALTER TABLE %s ALTER COLUMN %s ..." % (element.table.name, element.column.name)
+
+The second ``visit_alter_table`` will be invoked when any ``postgres`` dialect is used.
+
+The ``compiler`` argument is the :class:`~sqlalchemy.engine.base.Compiled` object
+in use. This object can be inspected for any information about the in-progress
+compilation, including ``compiler.dialect``, ``compiler.statement`` etc.
+The :class:`~sqlalchemy.sql.compiler.SQLCompiler` and :class:`~sqlalchemy.sql.compiler.DDLCompiler` (DDLCompiler is 0.6. only)
+both include a ``process()`` method which can be used for compilation of embedded attributes::
+
+ class InsertFromSelect(ClauseElement):
+ def __init__(self, table, select):
+ self.table = table
+ self.select = select
+
+ @compiles(InsertFromSelect)
+ def visit_insert_from_select(element, compiler, **kw):
+ return "INSERT INTO %s (%s)" % (
+ compiler.process(element.table, asfrom=True),
+ compiler.process(element.select)
+ )
+
+ insert = InsertFromSelect(t1, select([t1]).where(t1.c.x>5))
+ print insert
+
+Produces::
+
+ "INSERT INTO mytable (SELECT mytable.x, mytable.y, mytable.z FROM mytable WHERE mytable.x > :x_1)"
+
+
+"""
+
+def compiles(class_, *specs):
+ def decorate(fn):
+ existing = getattr(class_, '_compiler_dispatcher', None)
+ if not existing:
+ existing = _dispatcher()
+
+ # 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
+
+class _dispatcher(object):
+ def __init__(self):
+ self.specs = {}
+
+ def __call__(self, element, compiler, **kw):
+ # TODO: yes, this could also switch off of DBAPI in use.
+ fn = self.specs.get(compiler.dialect.name, None)
+ if not fn:
+ fn = self.specs['default']
+ return fn(element, compiler, **kw)
+
'ext.orderinglist',
'ext.associationproxy',
'ext.serializer',
+ 'ext.compiler',
)
if sys.version_info < (2, 4):
--- /dev/null
+import testenv; testenv.configure_for_tests()
+from sqlalchemy import *
+from sqlalchemy.sql.expression import ClauseElement, ColumnClause
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.sql import table, column
+from testlib import *
+
+class UserDefinedTest(TestBase, AssertsCompiledSQL):
+
+ def test_column(self):
+
+ class MyThingy(ColumnClause):
+ def __init__(self, arg= None):
+ super(MyThingy, self).__init__(arg or 'MYTHINGY!')
+
+ @compiles(MyThingy)
+ def visit_thingy(thingy, compiler, **kw):
+ return ">>%s<<" % thingy.name
+
+ self.assert_compile(
+ select([column('foo'), MyThingy()]),
+ "SELECT foo, >>MYTHINGY!<<"
+ )
+
+ self.assert_compile(
+ select([MyThingy('x'), MyThingy('y')]).where(MyThingy() == 5),
+ "SELECT >>x<<, >>y<< WHERE >>MYTHINGY!<< = :MYTHINGY!_1"
+ )
+
+ def test_stateful(self):
+ class MyThingy(ColumnClause):
+ def __init__(self):
+ super(MyThingy, self).__init__('MYTHINGY!')
+
+ @compiles(MyThingy)
+ def visit_thingy(thingy, compiler, **kw):
+ if not hasattr(compiler, 'counter'):
+ compiler.counter = 0
+ compiler.counter += 1
+ return str(compiler.counter)
+
+ self.assert_compile(
+ select([column('foo'), MyThingy()]).order_by(desc(MyThingy())),
+ "SELECT foo, 1 ORDER BY 2 DESC"
+ )
+
+ self.assert_compile(
+ select([MyThingy(), MyThingy()]).where(MyThingy() == 5),
+ "SELECT 1, 2 WHERE 3 = :MYTHINGY!_1"
+ )
+
+ def test_callout_to_compiler(self):
+ class InsertFromSelect(ClauseElement):
+ def __init__(self, table, select):
+ self.table = table
+ self.select = select
+
+ @compiles(InsertFromSelect)
+ def visit_insert_from_select(element, compiler, **kw):
+ return "INSERT INTO %s (%s)" % (
+ compiler.process(element.table, asfrom=True),
+ compiler.process(element.select)
+ )
+
+ t1 = table("mytable", column('x'), column('y'), column('z'))
+ self.assert_compile(
+ InsertFromSelect(
+ t1,
+ select([t1]).where(t1.c.x>5)
+ ),
+ "INSERT INTO mytable (SELECT mytable.x, mytable.y, mytable.z FROM mytable WHERE mytable.x > :x_1)"
+ )
+
+ def test_dialect_specific(self):
+ class AddThingy(ClauseElement):
+ __visit_name__ = 'add_thingy'
+
+ class DropThingy(ClauseElement):
+ __visit_name__ = 'drop_thingy'
+
+ @compiles(AddThingy, 'sqlite')
+ def visit_add_thingy(thingy, compiler, **kw):
+ return "ADD SPECIAL SL THINGY"
+
+ @compiles(AddThingy)
+ def visit_add_thingy(thingy, compiler, **kw):
+ return "ADD THINGY"
+
+ @compiles(DropThingy)
+ def visit_drop_thingy(thingy, compiler, **kw):
+ return "DROP THINGY"
+
+ self.assert_compile(AddThingy(),
+ "ADD THINGY"
+ )
+
+ self.assert_compile(DropThingy(),
+ "DROP THINGY"
+ )
+
+ from sqlalchemy.databases import sqlite as base
+ self.assert_compile(AddThingy(),
+ "ADD SPECIAL SL THINGY",
+ dialect=base.dialect()
+ )
+
+ self.assert_compile(DropThingy(),
+ "DROP THINGY",
+ dialect=base.dialect()
+ )
+
+ @compiles(DropThingy, 'sqlite')
+ def visit_drop_thingy(thingy, compiler, **kw):
+ return "DROP SPECIAL SL THINGY"
+
+ self.assert_compile(DropThingy(),
+ "DROP SPECIAL SL THINGY",
+ dialect=base.dialect()
+ )
+
+ self.assert_compile(DropThingy(),
+ "DROP THINGY",
+ )
+
+if __name__ == '__main__':
+ testenv.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
