def __repr__(self):
return "%s()" % self.__class__.__name__
- @property
- def bind(self):
- """Return the connectable associated with this SchemaItem."""
-
- m = self.metadata
- return m and m.bind or None
-
@property
def info(self):
try:
self.indexes = set()
self.constraints = set()
self._columns = expression.ColumnCollection()
- self.primary_key = PrimaryKeyConstraint()
+ self._set_primary_key(PrimaryKeyConstraint())
self._foreign_keys = util.OrderedSet()
self.ddl_listeners = util.defaultdict(list)
self.kwargs = {}
self._prefixes = kwargs.pop('prefixes', [])
- self.__extra_kwargs(**kwargs)
+ self._extra_kwargs(**kwargs)
# load column definitions from the database if 'autoload' is defined
# we do it after the table is in the singleton dictionary to support
# initialize all the column, etc. objects. done after reflection to
# allow user-overrides
- self.__post_init(*args, **kwargs)
+ self._init_items(*args)
def _init_existing(self, *args, **kwargs):
autoload = kwargs.pop('autoload', False)
if 'info' in kwargs:
self._info = kwargs.pop('info')
- self.__extra_kwargs(**kwargs)
- self.__post_init(*args, **kwargs)
+ self._extra_kwargs(**kwargs)
+ self._init_items(*args)
- def __extra_kwargs(self, **kwargs):
+ def _extra_kwargs(self, **kwargs):
# validate remaining kwargs that they all specify DB prefixes
if len([k for k in kwargs
if not re.match(r'^(?:%s)_' % '|'.join(dialects.__all__), k)]):
"Invalid argument(s) for Table: %s" % repr(kwargs.keys()))
self.kwargs.update(kwargs)
- def __post_init(self, *args, **kwargs):
- self._init_items(*args)
-
- @property
- def key(self):
- return _get_table_key(self.name, self.schema)
-
def _set_primary_key(self, pk):
if getattr(self, '_primary_key', None) in self.constraints:
self.constraints.remove(self._primary_key)
self._primary_key = pk
self.constraints.add(pk)
+ for c in pk.columns:
+ c.primary_key = True
+
+ @property
def primary_key(self):
return self._primary_key
- primary_key = property(primary_key, _set_primary_key)
def __repr__(self):
return "Table(%s)" % ', '.join(
def __str__(self):
return _get_table_key(self.description, self.schema)
+ @property
+ def bind(self):
+ """Return the connectable associated with this Table."""
+
+ m = self.metadata
+ return m and m.bind or None
+
def append_column(self, column):
"""Append a ``Column`` to this ``Table``."""
type_ = args.pop(0)
super(Column, self).__init__(name, None, type_)
- self.args = args
+ self._pending_args = args
self.key = kwargs.pop('key', name)
self.primary_key = kwargs.pop('primary_key', False)
self.nullable = kwargs.pop('nullable', not self.primary_key)
else:
return self.description
- @property
- def bind(self):
- return self.table.bind
-
def references(self, column):
"""Return True if this Column references the given column via foreign key."""
for fk in self.foreign_keys:
"before adding to a Table.")
if self.key is None:
self.key = self.name
- self.metadata = table.metadata
+
if getattr(self, 'table', None) is not None:
raise exc.ArgumentError("this Column already has a table!")
if self.key in table._columns:
# note the column being replaced, if any
self._pre_existing_column = table._columns.get(self.key)
+
table._columns.replace(self)
if self.primary_key:
- table.primary_key.replace(self)
+ table.primary_key._replace(self)
elif self.key in table.primary_key:
raise exc.ArgumentError(
"Trying to redefine primary-key column '%s' as a "
"non-primary-key column on table '%s'" % (
self.key, table.fullname))
- # if we think this should not raise an error, we'd instead do this:
- #table.primary_key.remove(self)
self.table = table
if self.index:
"external to the Table.")
table.append_constraint(UniqueConstraint(self.key))
- toinit = list(self.args)
+ toinit = list(self._pending_args)
if self.default is not None:
if isinstance(self.default, ColumnDefault):
toinit.append(self.default)
toinit.append(DefaultClause(self.server_onupdate,
for_update=True))
self._init_items(*toinit)
- self.args = None
+ del self._pending_args
def copy(self, **kw):
"""Create a copy of this ``Column``, unitialized.
This is used in ``Table.tometadata``.
"""
- return Column(self.name, self.type, self.default, key = self.key, primary_key = self.primary_key, nullable = self.nullable, quote=self.quote, index=self.index, autoincrement=self.autoincrement, *[c.copy(**kw) for c in self.constraints])
+ return Column(
+ self.name,
+ self.type,
+ self.default,
+ key = self.key,
+ primary_key = self.primary_key,
+ nullable = self.nullable,
+ quote=self.quote,
+ index=self.index,
+ autoincrement=self.autoincrement,
+ *[c.copy(**kw) for c in self.constraints])
def _make_proxy(self, selectable, name=None):
"""Create a *proxy* for this column.
__visit_name__ = 'foreign_key'
- def __init__(self, column, constraint=None, use_alter=False, name=None, onupdate=None, ondelete=None, deferrable=None, initially=None, link_to_name=False):
+ def __init__(self, column, constraint=None, use_alter=False, name=None, onupdate=None,
+ ondelete=None, deferrable=None, initially=None, link_to_name=False):
"""
Construct a column-level FOREIGN KEY.
def _set_parent(self, column):
if hasattr(self, 'parent'):
+ if self.parent is column:
+ return
raise exc.InvalidRequestError("This ForeignKey already has a parent !")
self.parent = column
self.constraint = ForeignKeyConstraint(
[], [], use_alter=self.use_alter, name=self.name,
onupdate=self.onupdate, ondelete=self.ondelete,
- deferrable=self.deferrable, initially=self.initially)
- self.parent.table.append_constraint(self.constraint)
- self.constraint._append_fk(self)
+ deferrable=self.deferrable, initially=self.initially,
+ )
+ self.constraint._elements[column] = self
+ self.constraint._set_parent(self.parent.table)
self.parent.foreign_keys.add(self)
self.parent.table.foreign_keys.add(self)
def __init__(self, for_update=False, metadata=None):
self.for_update = for_update
- self.metadata = util.assert_arg_type(metadata, (MetaData, type(None)), 'metadata')
def _set_parent(self, column):
self.column = column
- self.metadata = self.column.table.metadata
if self.for_update:
self.column.onupdate = self
else:
bind = _bind_or_error(self)
return bind._execute_default(self, **kwargs)
+ @property
+ def bind(self):
+ """Return the connectable associated with this default."""
+
+ return self.column.table.bind
+
def __repr__(self):
return "DefaultGenerator()"
# alias; deprecated starting 0.5.0
PassiveDefault = DefaultClause
-
class Constraint(SchemaItem):
- """A table-level SQL constraint, such as a KEY.
-
- Implements a hybrid of dict/setlike behavior with regards to the list of
- underying columns.
- """
+ """A table-level SQL constraint."""
__visit_name__ = 'constraint'
"""
self.name = name
- self.columns = expression.ColumnCollection()
self.deferrable = deferrable
self.initially = initially
+ @property
+ def table(self):
+ if isinstance(self.parent, Table):
+ return self.parent
+ else:
+ raise exc.InvalidRequestError("This constraint is not bound to a table.")
+
+ def _set_parent(self, parent):
+ self.parent = parent
+ parent.constraints.add(self)
+
+ def copy(self, **kw):
+ raise NotImplementedError()
+
+class ColumnCollectionConstraint(Constraint):
+ """A constraint that proxies a ColumnCollection."""
+
+ def __init__(self, *columns, **kw):
+ """
+ \*columns
+ A sequence of column names or Column objects.
+
+ name
+ Optional, the in-database name of this constraint.
+
+ deferrable
+ Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when
+ issuing DDL for this constraint.
+
+ initially
+ Optional string. If set, emit INITIALLY <value> when issuing DDL
+ for this constraint.
+
+ """
+ super(ColumnCollectionConstraint, self).__init__(**kw)
+ self.columns = expression.ColumnCollection()
+ self._pending_colargs = [_to_schema_column_or_string(c) for c in columns]
+ if self._pending_colargs and \
+ isinstance(self._pending_colargs[0], Column) and \
+ self._pending_colargs[0].table is not None:
+ self._set_parent(self._pending_colargs[0].table)
+
+ def _set_parent(self, table):
+ super(ColumnCollectionConstraint, self)._set_parent(table)
+ for col in self._pending_colargs:
+ if isinstance(col, basestring):
+ col = table.c[col]
+ self.columns.add(col)
+
def __contains__(self, x):
return x in self.columns
+ def copy(self, **kw):
+ return self.__class__(*self.columns.keys(),
+ name=self.name, deferrable=self.deferrable, initially=self.initially)
+
def contains_column(self, col):
return self.columns.contains_column(col)
- def keys(self):
- return self.columns.keys()
-
- def __add__(self, other):
- return self.columns + other
-
def __iter__(self):
return iter(self.columns)
def __len__(self):
return len(self.columns)
- def copy(self, **kw):
- raise NotImplementedError()
class CheckConstraint(Constraint):
"""A table- or column-level CHECK constraint.
Can be included in the definition of a Table or Column.
"""
- def __init__(self, sqltext, name=None, deferrable=None, initially=None):
+ def __init__(self, sqltext, name=None, deferrable=None, initially=None, table=None):
"""Construct a CHECK constraint.
sqltext
initially
Optional string. If set, emit INITIALLY <value> when issuing DDL
for this constraint.
+
"""
super(CheckConstraint, self).__init__(name, deferrable, initially)
raise exc.ArgumentError(
"sqltext must be a string and will be used verbatim.")
self.sqltext = sqltext
-
+ if table:
+ self._set_parent(table)
+
def __visit_name__(self):
if isinstance(self.parent, Table):
return "check_constraint"
return "column_check_constraint"
__visit_name__ = property(__visit_name__)
- def _set_parent(self, parent):
- self.parent = parent
- parent.constraints.add(self)
-
def copy(self, **kw):
return CheckConstraint(self.sqltext, name=self.name)
"""
__visit_name__ = 'foreign_key_constraint'
- def __init__(self, columns, refcolumns, name=None, onupdate=None, ondelete=None, use_alter=False, deferrable=None, initially=None, link_to_name=False):
+ def __init__(self, columns, refcolumns, name=None, onupdate=None, ondelete=None, deferrable=None, initially=None, use_alter=False, link_to_name=False, table=None):
"""Construct a composite-capable FOREIGN KEY.
:param columns: A sequence of local column names. The named columns must be defined
:param link_to_name: if True, the string name given in ``column`` is the rendered
name of the referenced column, not its locally assigned ``key``.
- :param use_alter: If True, do not emit this key as part of the CREATE TABLE
+ :param use_alter: If True, do not emit this constraint as part of the CREATE TABLE
definition. Instead, use ALTER TABLE after table creation to add
- the key. Useful for circular dependencies.
+ the key. Useful for circular dependencies and conditional constraint generation.
"""
super(ForeignKeyConstraint, self).__init__(name, deferrable, initially)
- self.__colnames = columns
- self.__refcolnames = refcolumns
- self.elements = util.OrderedSet()
+
self.onupdate = onupdate
self.ondelete = ondelete
self.link_to_name = link_to_name
if self.name is None and use_alter:
- raise exc.ArgumentError("Alterable ForeignKey/ForeignKeyConstraint requires a name")
+ raise exc.ArgumentError("Alterable Constraint requires a name")
self.use_alter = use_alter
- def _set_parent(self, table):
- self.table = table
- if self not in table.constraints:
- table.constraints.add(self)
- for (c, r) in zip(self.__colnames, self.__refcolnames):
- self.append_element(c, r)
+ self._elements = util.OrderedDict()
+ for col, refcol in zip(columns, refcolumns):
+ self._elements[col] = ForeignKey(
+ refcol,
+ constraint=self,
+ name=self.name,
+ onupdate=self.onupdate,
+ ondelete=self.ondelete,
+ use_alter=self.use_alter,
+ link_to_name=self.link_to_name
+ )
+
+ if table:
+ self._set_parent(table)
- def append_element(self, col, refcol):
- fk = ForeignKey(refcol, constraint=self, name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter, link_to_name=self.link_to_name)
- fk._set_parent(self.table.c[col])
- self._append_fk(fk)
-
- def _append_fk(self, fk):
- self.columns.add(self.table.c[fk.parent.key])
- self.elements.add(fk)
+ def _set_parent(self, table):
+ super(ForeignKeyConstraint, self)._set_parent(table)
+ for col, fk in self._elements.iteritems():
+ if isinstance(col, basestring):
+ col = table.c[col]
+ fk._set_parent(col)
def copy(self, **kw):
- return ForeignKeyConstraint([x.parent.name for x in self.elements], [x._get_colspec(**kw) for x in self.elements], name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter)
-
-class PrimaryKeyConstraint(Constraint):
+ return ForeignKeyConstraint(
+ [x.parent.name for x in self._elements.values()],
+ [x._get_colspec(**kw) for x in self._elements.values()],
+ name=self.name,
+ onupdate=self.onupdate,
+ ondelete=self.ondelete,
+ use_alter=self.use_alter
+ )
+
+class PrimaryKeyConstraint(ColumnCollectionConstraint):
"""A table-level PRIMARY KEY constraint.
Defines a single column or composite PRIMARY KEY constraint. For a
__visit_name__ = 'primary_key_constraint'
- def __init__(self, *columns, **kwargs):
- """Construct a composite-capable PRIMARY KEY.
-
- \*columns
- A sequence of column names. All columns named must be defined and
- present within the parent Table.
-
- name
- Optional, the in-database name of the key.
-
- deferrable
- Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when
- issuing DDL for this constraint.
-
- initially
- Optional string. If set, emit INITIALLY <value> when issuing DDL
- for this constraint.
- """
-
- constraint_args = dict(name=kwargs.pop('name', None),
- deferrable=kwargs.pop('deferrable', None),
- initially=kwargs.pop('initially', None))
- if kwargs:
- raise exc.ArgumentError(
- 'Unknown PrimaryKeyConstraint argument(s): %s' %
- ', '.join(repr(x) for x in kwargs.iterkeys()))
-
- super(PrimaryKeyConstraint, self).__init__(**constraint_args)
- self.__colnames = list(columns)
-
def _set_parent(self, table):
- self.table = table
- table.primary_key = self
- for name in self.__colnames:
- self.add(table.c[name])
-
- def add(self, col):
- self.columns.add(col)
- col.primary_key = True
- append_column = add
+ super(PrimaryKeyConstraint, self)._set_parent(table)
+ table._set_primary_key(self)
- def replace(self, col):
+ def _replace(self, col):
self.columns.replace(col)
- def remove(self, col):
- col.primary_key = False
- del self.columns[col.key]
-
- def copy(self, **kw):
- return PrimaryKeyConstraint(name=self.name, *[c.key for c in self])
-
- __hash__ = Constraint.__hash__
-
- def __eq__(self, other):
- return self.columns == other
-
-class UniqueConstraint(Constraint):
+class UniqueConstraint(ColumnCollectionConstraint):
"""A table-level UNIQUE constraint.
Defines a single column or composite UNIQUE constraint. For a no-frills,
__visit_name__ = 'unique_constraint'
- def __init__(self, *columns, **kwargs):
- """Construct a UNIQUE constraint.
-
- \*columns
- A sequence of column names. All columns named must be defined and
- present within the parent Table.
-
- name
- Optional, the in-database name of the key.
-
- deferrable
- Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when
- issuing DDL for this constraint.
-
- initially
- Optional string. If set, emit INITIALLY <value> when issuing DDL
- for this constraint.
- """
-
- constraint_args = dict(name=kwargs.pop('name', None),
- deferrable=kwargs.pop('deferrable', None),
- initially=kwargs.pop('initially', None))
- if kwargs:
- raise exc.ArgumentError(
- 'Unknown UniqueConstraint argument(s): %s' %
- ', '.join(repr(x) for x in kwargs.iterkeys()))
-
- super(UniqueConstraint, self).__init__(**constraint_args)
- self.__colnames = list(columns)
-
- def _set_parent(self, table):
- self.table = table
- table.constraints.add(self)
- for c in self.__colnames:
- self.append_column(table.c[c])
-
- def append_column(self, col):
- self.columns.add(col)
-
- def copy(self, **kw):
- return UniqueConstraint(name=self.name, *self.__colnames)
-
class Index(SchemaItem):
"""A table-level INDEX.
\*columns
Columns to include in the index. All columns must belong to the same
- table, and no column may appear more than once.
+ table.
\**kwargs
Keyword arguments include:
"""
self.name = name
- self.columns = []
+ self.columns = expression.ColumnCollection()
self.table = None
self.unique = kwargs.pop('unique', False)
self.kwargs = kwargs
def _init_items(self, *args):
for column in args:
- self.append_column(_to_schema_column(column))
+ column = _to_schema_column(column)
+ if self.table is None:
+ self._set_parent(column.table)
+ elif column.table != self.table:
+ # all columns muse be from same table
+ raise exc.ArgumentError(
+ "All index columns must be from same table. "
+ "%s is from %s not %s" % (column, column.table, self.table))
+ self.columns.add(column)
def _set_parent(self, table):
self.table = table
- self.metadata = table.metadata
table.indexes.add(self)
- def append_column(self, column):
- # make sure all columns are from the same table
- # and no column is repeated
- if self.table is None:
- self._set_parent(column.table)
- elif column.table != self.table:
- # all columns muse be from same table
- raise exc.ArgumentError(
- "All index columns must be from same table. "
- "%s is from %s not %s" % (column, column.table, self.table))
- elif column.name in [ c.name for c in self.columns ]:
- raise exc.ArgumentError(
- "A column may not appear twice in the "
- "same index (%s already has column %s)" % (self.name, column))
- self.columns.append(column)
+ @property
+ def bind(self):
+ """Return the connectable associated with this Index."""
+
+ return self.table.bind
def create(self, bind=None):
if bind is None:
bind = _bind_or_error(self)
bind.drop(self)
- def __str__(self):
- return repr(self)
-
def __repr__(self):
return 'Index("%s", %s%s)' % (self.name,
', '.join(repr(c) for c in self.columns),
# TODO: scan all other tables and remove FK _column
del self.tables[table.key]
- @util.deprecated('Deprecated. Use ``metadata.sorted_tables``')
- def table_iterator(self, reverse=True, tables=None):
- """Deprecated - use metadata.sorted_tables()."""
-
- from sqlalchemy.sql.util import sort_tables
- if tables is None:
- tables = self.tables.itervalues()
- else:
- tables = set(tables).intersection(self.tables.itervalues())
- ret = sort_tables(tables)
- if reverse:
- ret = reversed(ret)
- return iter(ret)
-
@property
def sorted_tables(self):
"""Returns a list of ``Table`` objects sorted in order of
"""
if bind is None:
bind = _bind_or_error(self)
+ # TODO!!! the listener stuff here needs to move to engine/ddl.py
for listener in self.ddl_listeners['before-create']:
listener('before-create', self, bind)
bind.create(self, checkfirst=checkfirst, tables=tables)
"""
if bind is None:
bind = _bind_or_error(self)
+ # TODO!!! the listener stuff here needs to move to engine/ddl.py
for listener in self.ddl_listeners['before-drop']:
listener('before-drop', self, bind)
bind.drop(self, checkfirst=checkfirst, tables=tables)
supports_execution = True
_autocommit = True
+ schema_item = None
+ on = None
+
+ def execute(self, bind=None, schema_item=None):
+ """Execute this DDL immediately.
+
+ Executes the DDL statement in isolation using the supplied
+ :class:`~sqlalchemy.engine.base.Connectable` or :class:`~sqlalchemy.engine.base.Connectable` assigned to the ``.bind`` property,
+ if not supplied. If the DDL has a conditional ``on`` criteria, it
+ will be invoked with None as the event.
+
+ bind
+ Optional, an ``Engine`` or ``Connection``. If not supplied, a
+ valid :class:`~sqlalchemy.engine.base.Connectable` must be present in the ``.bind`` property.
+
+ schema_item
+ Optional, defaults to None. Will be passed to the ``on`` callable
+ criteria, if any, and may provide string expansion data for the
+ statement. See ``execute_at`` for more information.
+ """
+
+ if bind is None:
+ bind = _bind_or_error(self)
+
+ if self._should_execute(None, schema_item, bind):
+ return bind.execute(self.against(schema_item))
+ else:
+ bind.engine.logger.info("DDL execution skipped, criteria not met.")
+
+ def execute_at(self, event, schema_item):
+ """Link execution of this DDL to the DDL lifecycle of a SchemaItem.
+
+ Links this ``DDL`` to a ``Table`` or ``MetaData`` instance, executing
+ it when that schema item is created or dropped. The DDL statement
+ will be executed using the same Connection and transactional context
+ as the Table create/drop itself. The ``.bind`` property of this
+ statement is ignored.
+
+ event
+ One of the events defined in the schema item's ``.ddl_events``;
+ e.g. 'before-create', 'after-create', 'before-drop' or 'after-drop'
+
+ schema_item
+ A Table or MetaData instance
+
+ When operating on Table events, the following additional ``statement``
+ string substitions are available::
+
+ %(table)s - the Table name, with any required quoting applied
+ %(schema)s - the schema name, with any required quoting applied
+ %(fullname)s - the Table name including schema, quoted if needed
+
+ The DDL's ``context``, if any, will be combined with the standard
+ substutions noted above. Keys present in the context will override
+ the standard substitutions.
+
+ A DDL instance can be linked to any number of schema items. The
+ statement subsitution support allows for DDL instances to be used in a
+ template fashion.
+
+ ``execute_at`` builds on the ``append_ddl_listener`` interface of
+ MetaDta and Table objects.
+
+ Caveat: Creating or dropping a Table in isolation will also trigger
+ any DDL set to ``execute_at`` that Table's MetaData. This may change
+ in a future release.
+ """
+
+ if not hasattr(schema_item, 'ddl_listeners'):
+ raise exc.ArgumentError(
+ "%s does not support DDL events" % type(schema_item).__name__)
+ if event not in schema_item.ddl_events:
+ raise exc.ArgumentError(
+ "Unknown event, expected one of (%s), got '%r'" %
+ (', '.join(schema_item.ddl_events), event))
+ schema_item.ddl_listeners[event].append(self)
+ return self
+
+ @expression._generative
+ def against(self, schema_item):
+ """Return a copy of this DDL against a specific schema item."""
+
+ self.schema_item = schema_item
+
+ def __call__(self, event, schema_item, bind):
+ """Execute the DDL as a ddl_listener."""
+
+ if self._should_execute(event, schema_item, bind):
+ return bind.execute(self.against(schema_item))
+
+ def _check_ddl_on(self, on):
+ if (on is not None and
+ (not isinstance(on, basestring) and not util.callable(on))):
+ raise exc.ArgumentError(
+ "Expected the name of a database dialect or a callable for "
+ "'on' criteria, got type '%s'." % type(on).__name__)
+
+ def _should_execute(self, event, schema_item, bind):
+ if self.on is None:
+ return True
+ elif isinstance(self.on, basestring):
+ return self.on == bind.engine.name
+ else:
+ return self.on(event, schema_item, bind)
+
def bind(self):
if self._bind:
return self._bind
raise exc.ArgumentError(
"Expected a string or unicode SQL statement, got '%r'" %
statement)
- if (on is not None and
- (not isinstance(on, basestring) and not util.callable(on))):
- raise exc.ArgumentError(
- "Expected the name of a database dialect or a callable for "
- "'on' criteria, got type '%s'." % type(on).__name__)
self.statement = statement
- self.on = on
self.context = context or {}
- self._bind = bind
- self.schema_item = None
-
- def execute(self, bind=None, schema_item=None):
- """Execute this DDL immediately.
-
- Executes the DDL statement in isolation using the supplied
- :class:`~sqlalchemy.engine.base.Connectable` or :class:`~sqlalchemy.engine.base.Connectable` assigned to the ``.bind`` property,
- if not supplied. If the DDL has a conditional ``on`` criteria, it
- will be invoked with None as the event.
-
- bind
- Optional, an ``Engine`` or ``Connection``. If not supplied, a
- valid :class:`~sqlalchemy.engine.base.Connectable` must be present in the ``.bind`` property.
-
- schema_item
- Optional, defaults to None. Will be passed to the ``on`` callable
- criteria, if any, and may provide string expansion data for the
- statement. See ``execute_at`` for more information.
- """
-
- if bind is None:
- bind = _bind_or_error(self)
- if self._should_execute(None, schema_item, bind):
- return bind.execute(self.against(schema_item))
- else:
- bind.engine.logger.info("DDL execution skipped, criteria not met.")
-
- def execute_at(self, event, schema_item):
- """Link execution of this DDL to the DDL lifecycle of a SchemaItem.
-
- Links this ``DDL`` to a ``Table`` or ``MetaData`` instance, executing
- it when that schema item is created or dropped. The DDL statement
- will be executed using the same Connection and transactional context
- as the Table create/drop itself. The ``.bind`` property of this
- statement is ignored.
-
- event
- One of the events defined in the schema item's ``.ddl_events``;
- e.g. 'before-create', 'after-create', 'before-drop' or 'after-drop'
-
- schema_item
- A Table or MetaData instance
-
- When operating on Table events, the following additional ``statement``
- string substitions are available::
-
- %(table)s - the Table name, with any required quoting applied
- %(schema)s - the schema name, with any required quoting applied
- %(fullname)s - the Table name including schema, quoted if needed
-
- The DDL's ``context``, if any, will be combined with the standard
- substutions noted above. Keys present in the context will override
- the standard substitutions.
-
- A DDL instance can be linked to any number of schema items. The
- statement subsitution support allows for DDL instances to be used in a
- template fashion.
-
- ``execute_at`` builds on the ``append_ddl_listener`` interface of
- MetaDta and Table objects.
-
- Caveat: Creating or dropping a Table in isolation will also trigger
- any DDL set to ``execute_at`` that Table's MetaData. This may change
- in a future release.
- """
-
- if not hasattr(schema_item, 'ddl_listeners'):
- raise exc.ArgumentError(
- "%s does not support DDL events" % type(schema_item).__name__)
- if event not in schema_item.ddl_events:
- raise exc.ArgumentError(
- "Unknown event, expected one of (%s), got '%r'" %
- (', '.join(schema_item.ddl_events), event))
- schema_item.ddl_listeners[event].append(self)
- return self
-
- @expression._generative
- def against(self, schema_item):
- """Return a copy of this DDL against a specific schema item."""
-
- self.schema_item = schema_item
-
- def __call__(self, event, schema_item, bind):
- """Execute the DDL as a ddl_listener."""
-
- if self._should_execute(event, schema_item, bind):
- return bind.execute(self.against(schema_item))
+ self._check_ddl_on(on)
+ self.on = on
+ self._bind = bind
- def _should_execute(self, event, schema_item, bind):
- if self.on is None:
- return True
- elif isinstance(self.on, basestring):
- return self.on == bind.engine.name
- else:
- return self.on(event, schema_item, bind)
def __repr__(self):
return '<%s@%s; %s>' % (
raise exc.ArgumentError("schema.Column object expected")
return element
+def _to_schema_column_or_string(element):
+ if hasattr(element, '__clause_element__'):
+ element = element.__clause_element__()
+ return element
+
class _CreateDropBase(DDLElement):
"""Base class for DDL constucts that represent CREATE and DROP or equivalents.
"""
- def __init__(self, element):
+ def __init__(self, element, on=None, bind=None):
self.element = element
+ self._check_ddl_on(on)
+ self.on = on
+ self.bind = bind
- def bind(self):
- if self._bind:
- return self._bind
- if self.element:
- e = self.element.bind
- if e:
- return e
- return None
-
- def _set_bind(self, bind):
- self._bind = bind
- bind = property(bind, _set_bind)
class CreateTable(_CreateDropBase):
"""Represent a CREATE TABLE statement."""
__visit_name__ = "drop_table"
-class AddForeignKey(_CreateDropBase):
- """Represent an ALTER TABLE ADD FOREIGN KEY statement."""
-
- __visit_name__ = "add_foreignkey"
-
-class DropForeignKey(_CreateDropBase):
- """Represent an ALTER TABLE DROP FOREIGN KEY statement."""
-
- __visit_name__ = "drop_foreignkey"
-
class CreateSequence(_CreateDropBase):
"""Represent a CREATE SEQUENCE statement."""
"""Represent a DROP INDEX statement."""
__visit_name__ = "drop_index"
+
+class AddConstraint(_CreateDropBase):
+ """Represent an ALTER TABLE ADD CONSTRAINT statement."""
+ __visit_name__ = "add_constraint"
+
+class DropConstraint(_CreateDropBase):
+ """Represent an ALTER TABLE DROP CONSTRAINT statement."""
+
+ __visit_name__ = "drop_constraint"
+
def _bind_or_error(schemaitem):
bind = schemaitem.bind
if not bind:
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
