class SchemaItem(visitors.Visitable):
"""Base class for items that define a database schema."""
+ __visit_name__ = 'schema_item'
quote = None
def _init_items(self, *args):
__metaclass__ = _TableSingleton
+ __visit_name__ = 'table'
+
ddl_events = ('before-create', 'after-create', 'before-drop', 'after-drop')
def __init__(self, name, metadata, *args, **kwargs):
class Column(SchemaItem, expression.ColumnClause):
"""Represents a column in a database table."""
+ __visit_name__ = 'column'
+
def __init__(self, *args, **kwargs):
"""
Construct a new ``Column`` object.
Further examples of foreign key configuration are in :ref:`metadata_foreignkeys`.
"""
+
+ __visit_name__ = 'foreign_key'
+
def __init__(self, column, constraint=None, use_alter=False, name=None, onupdate=None, ondelete=None, deferrable=None, initially=None):
"""
Construct a column-level FOREIGN KEY.
class DefaultGenerator(SchemaItem):
"""Base class for column *default* values."""
+ __visit_name__ = 'default_generator'
+
def __init__(self, for_update=False, metadata=None):
self.for_update = for_update
self.metadata = util.assert_arg_type(metadata, (MetaData, type(None)), 'metadata')
class Sequence(DefaultGenerator):
"""Represents a named database sequence."""
+ __visit_name__ = 'sequence'
+
def __init__(self, name, start=None, increment=None, schema=None,
optional=False, quote=None, **kwargs):
super(Sequence, self).__init__(**kwargs)
underying columns.
"""
+ __visit_name__ = 'constraint'
+
def __init__(self, name=None, deferrable=None, initially=None):
"""Create a SQL constraint.
Examples of foreign key configuration are in :ref:`metadata_foreignkeys`.
"""
+ __visit_name__ = 'foreign_key_constraint'
def __init__(self, columns, refcolumns, name=None, onupdate=None, ondelete=None, use_alter=False, deferrable=None, initially=None):
"""Construct a composite-capable FOREIGN KEY.
multiple-column PrimaryKeyConstraint.
"""
+ __visit_name__ = 'primary_key_constraint'
+
def __init__(self, *columns, **kwargs):
"""Construct a composite-capable PRIMARY KEY.
UniqueConstraint.
"""
+ __visit_name__ = 'unique_constraint'
+
def __init__(self, *columns, **kwargs):
"""Construct a UNIQUE constraint.
a shorthand equivalent for an unnamed, single column Index.
"""
+ __visit_name__ = 'index'
+
def __init__(self, name, *columns, **kwargs):
"""Construct an index object.
class ClauseElement(Visitable):
"""Base class for elements of a programmatically constructed SQL expression."""
+ __visit_name__ = 'clause'
+
_annotations = {}
supports_execution = False
_from_objects = []
"""
+ __visit_name__ = 'column'
primary_key = False
foreign_keys = []
quote = None
class Selectable(ClauseElement):
"""mark a class as being selectable"""
+ __visit_name__ = 'selectable'
class FromClause(Selectable):
"""Represent an element that can be used within the ``FROM`` clause of a ``SELECT`` statement."""
"""
+ __visit_name__ = 'null'
+
def __init__(self):
self.type = sqltypes.NULLTYPE
"""
+ __visit_name__ = 'function'
+
def __init__(self, name, *clauses, **kwargs):
self.packagenames = kwargs.get('packagenames', None) or []
self.name = name
class _Cast(ColumnElement):
+ __visit_name__ = 'cast'
+
def __init__(self, clause, totype, **kwargs):
self.type = sqltypes.to_instance(totype)
self.clause = _literal_as_binds(clause, None)
class _UnaryExpression(ColumnElement):
+
+ __visit_name__ = 'unary'
+
def __init__(self, element, operator=None, modifier=None, type_=None, negate=None):
self.operator = operator
self.modifier = modifier
class _BinaryExpression(ColumnElement):
"""Represent an expression that is ``LEFT <operator> RIGHT``."""
+ __visit_name__ = 'binary'
+
def __init__(self, left, right, operator, type_=None, negate=None, modifiers=None):
self.left = _literal_as_text(left).self_group(against=operator)
self.right = _literal_as_text(right).self_group(against=operator)
off all ``FromClause`` subclasses.
"""
+ __visit_name__ = 'join'
def __init__(self, left, right, onclause=None, isouter=False):
self.left = _selectable(left)
"""
+ __visit_name__ = 'alias'
named_with_column = True
def __init__(self, selectable, alias=None):
class _Grouping(ColumnElement):
"""Represent a grouping within a column expression"""
+ __visit_name__ = 'grouping'
+
def __init__(self, element):
self.element = element
self.type = getattr(element, 'type', None)
"""
+ __visit_name__ = 'label'
+
def __init__(self, name, element, type_=None):
while isinstance(element, _Label):
element = element.element
``ColumnClause``.
"""
+ __visit_name__ = 'column'
+
def __init__(self, text, selectable=None, type_=None, is_literal=False):
self.key = self.name = text
self.table = selectable
"""
+ __visit_name__ = 'table'
+
named_with_column = True
def __init__(self, name, *columns):
class _ScalarSelect(_Grouping):
- __visit_name__ = 'grouping'
_from_objects = []
def __init__(self, element):
class CompoundSelect(_SelectBaseMixin, FromClause):
"""Forms the basis of ``UNION``, ``UNION ALL``, and other SELECT-based set operations."""
+ __visit_name__ = 'compound_select'
+
def __init__(self, keyword, *selects, **kwargs):
self._should_correlate = kwargs.pop('correlate', False)
self.keyword = keyword
ability to execute themselves and return a result set.
"""
+
+ __visit_name__ = 'select'
+
def __init__(self, columns, whereclause=None, from_obj=None, distinct=False, having=None, correlate=True, prefixes=None, **kwargs):
"""Construct a Select object.
class _UpdateBase(ClauseElement):
"""Form the base for ``INSERT``, ``UPDATE``, and ``DELETE`` statements."""
+ __visit_name__ = 'update_base'
+
supports_execution = True
_autocommit = True
bind = property(bind, _set_bind)
class _ValuesBase(_UpdateBase):
+
+ __visit_name__ = 'values_base'
+
def __init__(self, table, values):
self.table = table
self.parameters = self._process_colparams(values)
The ``Insert`` object is created using the :func:`insert()` function.
"""
+ __visit_name__ = 'insert'
+
def __init__(self, table, values=None, inline=False, bind=None, prefixes=None, **kwargs):
_ValuesBase.__init__(self, table, values)
self._bind = bind
The ``Update`` object is created using the :func:`update()` function.
"""
+ __visit_name__ = 'update'
+
def __init__(self, table, whereclause, values=None, inline=False, bind=None, **kwargs):
_ValuesBase.__init__(self, table, values)
self._bind = bind
"""
+ __visit_name__ = 'delete'
+
def __init__(self, table, whereclause, bind=None, **kwargs):
self._bind = bind
self.table = table
self._whereclause = clone(self._whereclause)
class _IdentifiedClause(ClauseElement):
+ __visit_name__ = 'identified'
supports_execution = True
_autocommit = False
quote = None
self.ident = ident
class SavepointClause(_IdentifiedClause):
- pass
+ __visit_name__ = 'savepoint'
class RollbackToSavepointClause(_IdentifiedClause):
- pass
+ __visit_name__ = 'rollback_to_savepoint'
class ReleaseSavepointClause(_IdentifiedClause):
- pass
+ __visit_name__ = 'release_savepoint'
from sqlalchemy.sql.visitors import VisitableType
class _GenericMeta(VisitableType):
- def __init__(cls, clsname, bases, dict):
- cls.__visit_name__ = 'function'
- type.__init__(cls, clsname, bases, dict)
-
def __call__(self, *args, **kwargs):
args = [_literal_as_binds(c) for c in args]
return type.__call__(self, *args, **kwargs)
else:
return sql.and_(*crit)
-
+
class Annotated(object):
"""clones a ClauseElement and applies an 'annotations' dictionary.
try:
cls = annotated_classes[element.__class__]
except KeyError:
- raise KeyError("Class %s has not defined an Annotated subclass" % element.__class__)
+ cls = annotated_classes[element.__class__] = type.__new__(type,
+ "Annotated%s" % element.__class__.__name__,
+ (Annotated, element.__class__), {})
return object.__new__(cls)
def __init__(self, element, values):
'cloned_traverse', 'replacement_traverse']
class VisitableType(type):
- """Metaclass which applies a `__visit_name__` attribute and
- `_compiler_dispatch` method to classes.
+ """Metaclass which checks for a `__visit_name__` attribute and
+ applies `_compiler_dispatch` method to classes.
"""
- def __init__(cls, clsname, bases, dict):
- if not '__visit_name__' in cls.__dict__:
- m = re.match(r'_?(\w+?)(?:Expression|Clause|Element|$)', clsname)
- x = m.group(1)
- x = re.sub(r'(?!^)[A-Z]', lambda m:'_'+m.group(0).lower(), x)
- cls.__visit_name__ = x.lower()
+ def __init__(cls, clsname, bases, clsdict):
+ if cls.__name__ == 'Visitable':
+ super(VisitableType, cls).__init__(clsname, bases, clsdict)
+ return
+
+ assert hasattr(cls, '__visit_name__'), "`Visitable` descendants " \
+ "should define `__visit_name__`"
# set up an optimized visit dispatch function
# for use by the compiler
- visit_name = cls.__dict__["__visit_name__"]
+ visit_name = cls.__visit_name__
if isinstance(visit_name, str):
getter = operator.attrgetter("visit_%s" % visit_name)
def _compiler_dispatch(self, visitor, **kw):
cls._compiler_dispatch = _compiler_dispatch
- super(VisitableType, cls).__init__(clsname, bases, dict)
+ super(VisitableType, cls).__init__(clsname, bases, clsdict)
class Visitable(object):
"""Base class for visitable objects, applies the
# establish two ficticious ClauseElements.
# define deep equality semantics as well as deep identity semantics.
class A(ClauseElement):
+ __visit_name__ = 'a'
+
def __init__(self, expr):
self.expr = expr
return "A(%s)" % repr(self.expr)
class B(ClauseElement):
+ __visit_name__ = 'b'
+
def __init__(self, *items):
self.items = items
assert struct != s3
assert struct3 == s3
+ def test_visit_name(self):
+ # override fns in testlib/schema.py
+ from sqlalchemy import Column
+
+ class CustomObj(Column):
+ pass
+
+ assert CustomObj.__visit_name__ == Column.__visit_name__ == 'column'
+
+ foo, bar = CustomObj('foo', String), CustomObj('bar', String)
+ bin = foo == bar
+ s = set(ClauseVisitor().iterate(bin))
+ assert set(ClauseVisitor().iterate(bin)) == set([foo, bar, bin])
class ClauseTest(TestBase, AssertsCompiledSQL):
"""test copy-in-place behavior of various ClauseElements."""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
