return schema + "." + name
class DDLEvents(event.Events):
+ """
+ Define create/drop event listers for schema objects.
+
+ See also:
+
+ :mod:`sqlalchemy.event`
+
+ """
+
def on_before_create(self, target, connection, **kw):
pass
target = None
on = None
+ dialect = None
+ callable_ = None
def execute(self, bind=None, target=None):
"""Execute this DDL immediately.
if bind is None:
bind = _bind_or_error(self)
- if self._should_execute(None, target, bind):
+ if self._should_execute(target, bind):
return bind.execute(self.against(target))
else:
bind.engine.logger.info(
def execute_at(self, event_name, target):
"""Link execution of this DDL to the DDL lifecycle of a SchemaItem.
-
- Links this ``DDLElement`` 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.
- :param event_name:
- Name of an event from :class:`.DDLEvents`. e.g.:
- 'on_before_create', 'on_after_create', 'on_before_drop',
- 'on_after_drop'.
-
- :param target:
- The Table or MetaData instance for which this DDLElement will
- be associated with.
-
- A DDLElement instance can be linked to any number of schema items.
-
- ``execute_at`` builds on the ``append_ddl_listener`` interface of
- :class:`MetaData` and :class:`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.
+ Deprecated. See :class:`.DDLEvents`, as well as
+ :meth:`.DDLEvent.execute_if`.
+
"""
- event_name = "on_" + event_name.replace('-', '_')
def call_event(target, connection, **kw):
- self(event_name, target, connection, **kw)
+ if self._should_execute_deprecated(event_name,
+ target, connection, **kw):
+ return connection.execute(self.against(target))
- event.listen(call_event, event_name, target)
+ event.listen(call_event, "on_" + event_name.replace('-', '_'), target)
@expression._generative
def against(self, target):
self.target = target
- def __call__(self, event, target, bind, **kw):
- """Execute the DDL as a ddl_listener."""
+ @expression._generative
+ def execute_if(self, dialect=None, callable_=None):
+ """Return a callable that will execute this
+ DDLElement conditionally.
+
+ Used to provide a wrapper for event listening::
+
+ event.listen(
+ DDL("my_ddl").execute_if(dialect='postgresql'),
+ 'on_before_create',
+ metadata
+ )
+
+ See also:
+
+ :class:`.DDLEvents`
+ :mod:`sqlalchemy.event`
+
+ """
+ self.dialect = dialect
+ self.callable_ = callable_
- if self._should_execute(event, target, bind, **kw):
+ def _should_execute(self, target, bind, **kw):
+ if self.on is not None and \
+ not self._should_execute_deprecated(None, target, bind, **kw):
+ return False
+
+ if isinstance(self.dialect, basestring):
+ if self.dialect != bind.engine.name:
+ return False
+ elif isinstance(self.dialect, (tuple, list, set)):
+ if bind.engine.name not in self.dialect:
+ return False
+ if self.callable_ is not None and \
+ not self.callable_(self, target, bind, **kw):
+ return False
+
+ return True
+
+ def _should_execute_deprecated(self, event, target, bind, **kw):
+ if self.on is None:
+ return True
+ elif isinstance(self.on, basestring):
+ return self.on == bind.engine.name
+ elif isinstance(self.on, (tuple, list, set)):
+ return bind.engine.name in self.on
+ else:
+ return self.on(self, event, target, bind, **kw)
+
+ def __call__(self, target, bind, **kw):
+ """Execute the DDL as a ddl_listener."""
+
+ if self._should_execute(target, bind, **kw):
return bind.execute(self.against(target))
def _check_ddl_on(self, on):
"of names, or a callable for "
"'on' criteria, got type '%s'." % type(on).__name__)
- def _should_execute(self, event, target, bind, **kw):
- if self.on is None:
- return True
- elif isinstance(self.on, basestring):
- return self.on == bind.engine.name
- elif isinstance(self.on, (tuple, list, set)):
- return bind.engine.name in self.on
- else:
- return self.on(self, event, target, bind, **kw)
-
def bind(self):
if self._bind:
return self._bind
SQL bind parameters are not available in DDL statements.
:param on:
- Optional filtering criteria. May be a string, tuple or a callable
- predicate. If a string, it will be compared to the name of the
- executing database dialect::
-
- DDL('something', on='postgresql')
-
- If a tuple, specifies multiple dialect names::
-
- DDL('something', on=('postgresql', 'mysql'))
-
- If a callable, it will be invoked with four positional arguments
- as well as optional keyword arguments:
-
- :ddl:
- This DDL element.
-
- :event:
- The name of the event that has triggered this DDL, such as
- 'on_after_create' Will be None if the DDL is executed
- explicitly.
-
- :target:
- The ``Table`` or ``MetaData`` object which is the target of
- this event. May be None if the DDL is executed explicitly.
-
- :connection:
- The ``Connection`` being used for DDL execution
-
- :tables:
- Optional keyword argument - a list of Table objects which are to
- be created/ dropped within a MetaData.create_all() or drop_all()
- method call.
-
-
- If the callable returns a true value, the DDL statement will be
- executed.
+ Deprecated. See :meth:`.DDLElement.execute_if`.
:param context:
Optional dictionary, defaults to None. These values will be
:param bind:
Optional. A :class:`~sqlalchemy.engine.base.Connectable`, used by
default when ``execute()`` is invoked without a bind argument.
+
+
+ See also:
+
+ :class:`.DDLEvents`
+ :mod:`sqlalchemy.event`
"""
def test_table_by_metadata(self):
metadata, users, engine = self.metadata, self.users, self.engine
+
+ event.listen(DDL('mxyzptlk'), 'on_before_create', users)
+ event.listen(DDL('klptzyxm'), 'on_after_create', users)
+ event.listen(DDL('xyzzy'), 'on_before_drop', users)
+ event.listen(DDL('fnord'), 'on_after_drop', users)
+
+ metadata.create_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' in strings
+ assert 'klptzyxm' in strings
+ assert 'xyzzy' not in strings
+ assert 'fnord' not in strings
+ del engine.mock[:]
+ metadata.drop_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' not in strings
+ assert 'klptzyxm' not in strings
+ assert 'xyzzy' in strings
+ assert 'fnord' in strings
+
+ def test_table_by_metadata_deprecated(self):
+ metadata, users, engine = self.metadata, self.users, self.engine
DDL('mxyzptlk').execute_at('before-create', users)
DDL('klptzyxm').execute_at('after-create', users)
DDL('xyzzy').execute_at('before-drop', users)
assert 'klptzyxm' not in strings
assert 'xyzzy' in strings
assert 'fnord' in strings
-
+
+
+ def test_metadata(self):
+ metadata, engine = self.metadata, self.engine
+
+ event.listen(DDL('mxyzptlk'), 'on_before_create', metadata)
+ event.listen(DDL('klptzyxm'), 'on_after_create', metadata)
+ event.listen(DDL('xyzzy'), 'on_before_drop', metadata)
+ event.listen(DDL('fnord'), 'on_after_drop', metadata)
+
+ metadata.create_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' in strings
+ assert 'klptzyxm' in strings
+ assert 'xyzzy' not in strings
+ assert 'fnord' not in strings
+ del engine.mock[:]
+ metadata.drop_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' not in strings
+ assert 'klptzyxm' not in strings
+ assert 'xyzzy' in strings
+ assert 'fnord' in strings
+
+ def test_metadata_deprecated(self):
+ metadata, engine = self.metadata, self.engine
+
+ DDL('mxyzptlk').execute_at('before-create', metadata)
+ DDL('klptzyxm').execute_at('after-create', metadata)
+ DDL('xyzzy').execute_at('before-drop', metadata)
+ DDL('fnord').execute_at('after-drop', metadata)
+
+ metadata.create_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' in strings
+ assert 'klptzyxm' in strings
+ assert 'xyzzy' not in strings
+ assert 'fnord' not in strings
+ del engine.mock[:]
+ metadata.drop_all()
+ strings = [str(x) for x in engine.mock]
+ assert 'mxyzptlk' not in strings
+ assert 'klptzyxm' not in strings
+ assert 'xyzzy' in strings
+ assert 'fnord' in strings
+
def test_conditional_constraint(self):
metadata, users, engine = self.metadata, self.users, self.engine
nonpg_mock = engines.mock_engine(dialect_name='sqlite')
# by placing the constraint in an Add/Drop construct, the
# 'inline_ddl' flag is set to False
+ event.listen(
+ AddConstraint(constraint).execute_if(dialect='postgresql'),
+ 'on_after_create',
+ users
+ )
+
+ event.listen(
+ DropConstraint(constraint).execute_if(dialect='postgresql'),
+ 'on_before_drop',
+ users
+ )
+
+ metadata.create_all(bind=nonpg_mock)
+ strings = ' '.join(str(x) for x in nonpg_mock.mock)
+ assert 'my_test_constraint' not in strings
+ metadata.drop_all(bind=nonpg_mock)
+ strings = ' '.join(str(x) for x in nonpg_mock.mock)
+ assert 'my_test_constraint' not in strings
+ metadata.create_all(bind=pg_mock)
+ strings = ' '.join(str(x) for x in pg_mock.mock)
+ assert 'my_test_constraint' in strings
+ metadata.drop_all(bind=pg_mock)
+ strings = ' '.join(str(x) for x in pg_mock.mock)
+ assert 'my_test_constraint' in strings
+
+ def test_conditional_constraint_deprecated(self):
+ metadata, users, engine = self.metadata, self.users, self.engine
+ nonpg_mock = engines.mock_engine(dialect_name='sqlite')
+ pg_mock = engines.mock_engine(dialect_name='postgresql')
+ constraint = CheckConstraint('a < b', name='my_test_constraint'
+ , table=users)
+
+ # by placing the constraint in an Add/Drop construct, the
+ # 'inline_ddl' flag is set to False
+
AddConstraint(constraint, on='postgresql'
).execute_at('after-create', users)
DropConstraint(constraint, on='postgresql'
metadata.drop_all(bind=pg_mock)
strings = ' '.join(str(x) for x in pg_mock.mock)
assert 'my_test_constraint' in strings
-
- def test_metadata(self):
- metadata, engine = self.metadata, self.engine
- DDL('mxyzptlk').execute_at('before-create', metadata)
- DDL('klptzyxm').execute_at('after-create', metadata)
- DDL('xyzzy').execute_at('before-drop', metadata)
- DDL('fnord').execute_at('after-drop', metadata)
-
- metadata.create_all()
- strings = [str(x) for x in engine.mock]
- assert 'mxyzptlk' in strings
- assert 'klptzyxm' in strings
- assert 'xyzzy' not in strings
- assert 'fnord' not in strings
- del engine.mock[:]
- metadata.drop_all()
- strings = [str(x) for x in engine.mock]
- assert 'mxyzptlk' not in strings
- assert 'klptzyxm' not in strings
- assert 'xyzzy' in strings
- assert 'fnord' in strings
def test_ddl_execute(self):
try:
self.assert_compile(ddl.against(insane_schema),
'S S-T T-"s s"."t t"-b', dialect=dialect)
+
def test_filter(self):
cx = self.mock_engine()
tbl = Table('t', MetaData(), Column('id', Integer))
target = cx.name
- assert DDL('')._should_execute('x', tbl, cx)
- assert DDL('', on=target)._should_execute('x', tbl, cx)
- assert not DDL('', on='bogus')._should_execute('x', tbl, cx)
- assert DDL('', on=lambda d, x,y,z: True)._should_execute('x', tbl, cx)
+ assert DDL('')._should_execute(tbl, cx)
+ assert DDL('').execute_if(dialect=target)._should_execute(tbl, cx)
+ assert not DDL('').execute_if(dialect='bogus').\
+ _should_execute(tbl, cx)
+ assert DDL('').execute_if(callable_=lambda d, y,z: True).\
+ _should_execute(tbl, cx)
+ assert(DDL('').execute_if(
+ callable_=lambda d, y,z: z.engine.name
+ != 'bogus').
+ _should_execute(tbl, cx))
+
+ def test_filter_deprecated(self):
+ cx = self.mock_engine()
+
+ tbl = Table('t', MetaData(), Column('id', Integer))
+ target = cx.name
+
+ assert DDL('')._should_execute_deprecated('x', tbl, cx)
+ assert DDL('', on=target)._should_execute_deprecated('x', tbl, cx)
+ assert not DDL('', on='bogus').\
+ _should_execute_deprecated('x', tbl, cx)
+ assert DDL('', on=lambda d, x,y,z: True).\
+ _should_execute_deprecated('x', tbl, cx)
assert(DDL('', on=lambda d, x,y,z: z.engine.name != 'bogus').
- _should_execute('x', tbl, cx))
+ _should_execute_deprecated('x', tbl, cx))
def test_repr(self):
assert repr(DDL('s'))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
