.. autoclass:: sqlalchemy.events.DDLEvents
:members:
+.. autoclass:: sqlalchemy.events.SchemaEventTarget
+ :members:
+
:undoc-members:
:show-inheritance:
+.. autoclass:: SchemaItem
+ :show-inheritance:
+
.. autoclass:: Table
:members:
:undoc-members:
:undoc-members:
:show-inheritance:
+
.. _metadata_reflection:
Reflecting Database Objects
# supports_sane_rowcount.
client_flag = opts.get('client_flag', 0)
if self.dbapi is not None:
- CLIENT_FLAGS = __import__(
- self.dbapi.__name__ + '.constants.CLIENT'
- ).constants.CLIENT
- client_flag |= CLIENT_FLAGS.FOUND_ROWS
+ try:
+ CLIENT_FLAGS = __import__(
+ self.dbapi.__name__ + '.constants.CLIENT'
+ ).constants.CLIENT
+ client_flag |= CLIENT_FLAGS.FOUND_ROWS
+ except (AttributeError, ImportError):
+ pass
opts['client_flag'] = client_flag
return [[], opts]
"""
def __call__(self, _parent_cls):
- return _parent_cls.__dict__['dispatch'].dispatch_cls(_parent_cls)
+ for cls in _parent_cls.__mro__:
+ if 'dispatch' in cls.__dict__:
+ return cls.__dict__['dispatch'].dispatch_cls(_parent_cls)
+ else:
+ raise AttributeError("No class with a 'dispatch' member present.")
class _Dispatch(object):
"""Mirror the event listening definitions of an Events class with
assert isinstance(target, type), \
"Class-level Event targets must be classes."
- for cls in [target] + target.__subclasses__():
+ stack = [target]
+ while stack:
+ cls = stack.pop(0)
+ stack.extend(cls.__subclasses__())
self._clslevel[cls].append(obj)
def remove(self, obj, target):
- for cls in [target] + target.__subclasses__():
+ stack = [target]
+ while stack:
+ cls = stack.pop(0)
+ stack.extend(cls.__subclasses__())
self._clslevel[cls].remove(obj)
def clear(self):
class DDLEvents(event.Events):
"""
- Define create/drop event listers for schema objects.
-
- These events currently apply to :class:`.Table`
- and :class:`.MetaData` objects as targets.
-
- e.g.::
+ Define event listeners for schema objects,
+ that is, :class:`.SchemaItem` and :class:`.SchemaEvent`
+ subclasses, including :class:`.MetaData`, :class:`.Table`,
+ :class:`.Column`.
+
+ :class:`.MetaData` and :class:`.Table` support events
+ specifically regarding when CREATE and DROP
+ DDL is emitted to the database.
+
+ Attachment events are also provided to customize
+ behavior whenever a child schema element is associated
+ with a parent, such as, when a :class:`.Column` is associated
+ with its :class:`.Table`, when a :class:`.ForeignKeyConstraint`
+ is associated with a :class:`.Table`, etc.
+
+ Example using the ``after_create`` event::
from sqlalchemy import event
from sqlalchemy import Table, Column, Metadata, Integer
"""
+ def before_parent_attach(self, target, parent):
+ """Called before a :class:`.SchemaItem` is associated with
+ a parent :class:`.SchemaItem`.
+
+ """
+
+ def after_parent_attach(self, target, parent):
+ """Called after a :class:`.SchemaItem` is associated with
+ a parent :class:`.SchemaItem`.
+
+ """
+
+class SchemaEventTarget(object):
+ """Base class for elements that are the targets of :class:`.DDLEvents` events.
+
+ This includes :class:`.SchemaItem` as well as :class:`.SchemaType`.
+
+ """
+ dispatch = event.dispatcher(DDLEvents)
+
+ def _set_parent(self, parent):
+ """Associate with this SchemaEvent's parent object."""
+
+ raise NotImplementedError()
+
+ def _set_parent_with_dispatch(self, parent):
+ self.dispatch.before_parent_attach(self, parent)
+ self._set_parent(parent)
+ self.dispatch.after_parent_attach(self, parent)
class PoolEvents(event.Events):
"""Available events for :class:`.Pool`.
RETAIN_SCHEMA = util.symbol('retain_schema')
-class SchemaItem(visitors.Visitable):
+class SchemaItem(events.SchemaEventTarget, visitors.Visitable):
"""Base class for items that define a database schema."""
__visit_name__ = 'schema_item'
for item in args:
if item is not None:
- item._set_parent(self)
-
- def _set_parent(self, parent):
- """Associate with this SchemaItem's parent object."""
-
- raise NotImplementedError()
+ item._set_parent_with_dispatch(self)
def get_children(self, **kwargs):
"""used to allow SchemaVisitor access"""
__visit_name__ = 'table'
- dispatch = event.dispatcher(events.DDLEvents)
-
def __new__(cls, *args, **kw):
if not args:
# python3k pickle seems to call this
raise exc.InvalidRequestError(
"Table '%s' not defined" % (key))
table = object.__new__(cls)
+ table.dispatch.before_parent_attach(table, metadata)
metadata._add_table(name, schema, table)
try:
table._init(name, metadata, *args, **kw)
+ table.dispatch.after_parent_attach(table, metadata)
return table
except:
metadata._remove_table(name, schema)
def append_column(self, column):
"""Append a ``Column`` to this ``Table``."""
- column._set_parent(self)
+ column._set_parent_with_dispatch(self)
def append_constraint(self, constraint):
"""Append a ``Constraint`` to this ``Table``."""
- constraint._set_parent(self)
+ constraint._set_parent_with_dispatch(self)
def append_ddl_listener(self, event_name, listener):
"""Append a DDL event listener to this ``Table``.
self.autoincrement = kwargs.pop('autoincrement', True)
self.constraints = set()
self.foreign_keys = set()
- self._table_events = set()
# check if this Column is proxying another column
if '_proxies' in kwargs:
self.proxies = kwargs.pop('_proxies')
# otherwise, add DDL-related events
elif isinstance(self.type, types.SchemaType):
- self.type._set_parent(self)
+ self.type._set_parent_with_dispatch(self)
if self.default is not None:
if isinstance(self.default, (ColumnDefault, Sequence)):
return False
def append_foreign_key(self, fk):
- fk._set_parent(self)
+ fk._set_parent_with_dispatch(self)
def __repr__(self):
kwarg = []
"Index object external to the Table.")
table.append_constraint(UniqueConstraint(self.key))
- for fn in self._table_events:
- fn(table, self)
- del self._table_events
-
def _on_table_attach(self, fn):
if self.table is not None:
- fn(self.table, self)
- else:
- self._table_events.add(fn)
+ fn(self, self.table)
+ event.listen(self, 'after_parent_attach', fn)
def copy(self, **kw):
"""Create a copy of this ``Column``, unitialized.
[c.copy(**kw) for c in self.constraints] + \
[c.copy(**kw) for c in self.foreign_keys if not c.constraint]
- c = Column(
+ return Column(
name=self.name,
type_=self.type,
key = self.key,
doc=self.doc,
*args
)
- if hasattr(self, '_table_events'):
- c._table_events = list(self._table_events)
- return c
def _make_proxy(self, selectable, name=None):
"""Create a *proxy* for this column.
selectable._columns.add(c)
if self.primary_key:
selectable.primary_key.add(c)
- for fn in c._table_events:
- fn(selectable, c)
- del c._table_events
+ c.dispatch.after_parent_attach(c, selectable)
return c
def get_children(self, schema_visitor=False, **kwargs):
self.parent.foreign_keys.add(self)
self.parent._on_table_attach(self._set_table)
- def _set_table(self, table, column):
+ def _set_table(self, column, table):
# standalone ForeignKey - create ForeignKeyConstraint
# on the hosting Table when attached to the Table.
if self.constraint is None and isinstance(table, Table):
deferrable=self.deferrable, initially=self.initially,
)
self.constraint._elements[self.parent] = self
- self.constraint._set_parent(table)
+ self.constraint._set_parent_with_dispatch(table)
table.foreign_keys.add(self)
class DefaultGenerator(SchemaItem):
super(Sequence, self)._set_parent(column)
column._on_table_attach(self._set_table)
- def _set_table(self, table, column):
+ def _set_table(self, column, table):
self.metadata = table.metadata
@property
bind.drop(self, checkfirst=checkfirst)
-class FetchedValue(object):
+class FetchedValue(events.SchemaEventTarget):
"""A marker for a transparent database-side default.
Use :class:`.FetchedValue` when the database is configured
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)
+ self._set_parent_with_dispatch(self._pending_colargs[0].table)
def _set_parent(self, table):
for col in self._pending_colargs:
__init__(name, deferrable, initially, _create_rule)
self.sqltext = expression._literal_as_text(sqltext)
if table is not None:
- self._set_parent(table)
+ self._set_parent_with_dispatch(table)
def __visit_name__(self):
if isinstance(self.parent, Table):
)
if table is not None:
- self._set_parent(table)
+ self._set_parent_with_dispatch(table)
@property
def columns(self):
# resolved to Column objects
if isinstance(col, basestring):
col = table.c[col]
- fk._set_parent(col)
+
+ if not hasattr(fk, 'parent') or \
+ fk.parent is not col:
+ fk._set_parent_with_dispatch(col)
if self.use_alter:
def supports_alter(ddl, event, schema_item, bind, **kw):
__visit_name__ = 'metadata'
- dispatch = event.dispatcher(events.DDLEvents)
-
def __init__(self, bind=None, reflect=False):
"""Create a new MetaData object.
from sqlalchemy.util.compat import decimal
from sqlalchemy.sql.visitors import Visitable
from sqlalchemy import util
-from sqlalchemy import processors
+from sqlalchemy import processors, events
import collections
default = util.importlater("sqlalchemy.engine", "default")
'LargeBinary.')
LargeBinary.__init__(self, *arg, **kw)
-class SchemaType(object):
+class SchemaType(events.SchemaEventTarget):
"""Mark a type as possibly requiring schema-level DDL for usage.
Supports types that must be explicitly created/dropped (i.e. PG ENUM type)
as well as types that are complimented by table or schema level
constraints, triggers, and other rules.
+
+ :class:`.SchemaType` classes can also be targets for the
+ :meth:`.DDLEvents.before_parent_attach` and :meth:`.DDLEvents.after_parent_attach`
+ events, where the events fire off surrounding the association of
+ the type object with a parent :class:`.Column`.
"""
def _set_parent(self, column):
column._on_table_attach(util.portable_instancemethod(self._set_table))
- def _set_table(self, table, column):
+ def _set_table(self, column, table):
table.append_ddl_listener('before-create',
util.portable_instancemethod(
self._on_table_create))
return not self.native_enum or \
not compiler.dialect.supports_native_enum
- def _set_table(self, table, column):
+ def _set_table(self, column, table):
if self.native_enum:
- SchemaType._set_table(self, table, column)
+ SchemaType._set_table(self, column, table)
e = schema.CheckConstraint(
def _should_create_constraint(self, compiler):
return not compiler.dialect.supports_native_boolean
- def _set_table(self, table, column):
+ def _set_table(self, column, table):
if not self.create_constraint:
return
metadata = MetaData(engine)
engine.connect()
- @profiling.function_call_count(3012, {'2.4': 1711})
+ @profiling.function_call_count(3266, {'2.4': 1711})
def test_profile_1_create_tables(self):
self.test_baseline_1_create_tables()
session = sessionmaker()()
engine.connect()
- @profiling.function_call_count(4592)
+ @profiling.function_call_count(4912)
def test_profile_1_create_tables(self):
self.test_baseline_1_create_tables()
import pickle
from sqlalchemy import Integer, String, UniqueConstraint, \
CheckConstraint, ForeignKey, MetaData, Sequence, \
- ForeignKeyConstraint, ColumnDefault, Index
+ ForeignKeyConstraint, ColumnDefault, Index, event,\
+ events
from test.lib.schema import Table, Column
from sqlalchemy import schema, exc
import sqlalchemy as tsa
def test_uninitialized_column_copy_events(self):
msgs = []
- def write(t, c):
+ def write(c, t):
msgs.append("attach %s.%s" % (t.name, c.name))
c1 = Column('foo', String())
- c1._on_table_attach(write)
m = MetaData()
for i in xrange(3):
cx = c1.copy()
+ # as of 0.7, these events no longer copy. its expected
+ # that listeners will be re-established from the
+ # natural construction of things.
+ cx._on_table_attach(write)
t = Table('foo%d' % i, m, cx)
eq_(msgs, ['attach foo0.foo', 'attach foo1.foo', 'attach foo2.foo'])
c.info['bar'] = 'zip'
assert c.info['bar'] == 'zip'
+
+class CatchAllEventsTest(TestBase):
+
+ def teardown(self):
+ events.SchemaEventTarget.dispatch._clear()
+
+ def test_all_events(self):
+ canary = []
+ def before_attach(obj, parent):
+ canary.append("%s->%s" % (obj.__class__.__name__, parent.__class__.__name__))
+
+ def after_attach(obj, parent):
+ canary.append("%s->%s" % (obj, parent))
+
+ event.listen(events.SchemaEventTarget, "before_parent_attach", before_attach)
+ event.listen(events.SchemaEventTarget, "after_parent_attach", after_attach)
+
+ m = MetaData()
+ t1 = Table('t1', m,
+ Column('id', Integer, Sequence('foo_id'), primary_key=True),
+ Column('bar', String, ForeignKey('t2.id'))
+ )
+ t2 = Table('t2', m,
+ Column('id', Integer, primary_key=True),
+ )
+
+ # TODO: test more conditions here, constraints, defaults, etc.
+ eq_(
+ canary,
+ [
+ 'Sequence->Column',
+ "Sequence('foo_id', start=None, increment=None, optional=False)->id",
+ 'ForeignKey->Column',
+ "ForeignKey('t2.id')->bar",
+ 'Table->MetaData',
+ 'Column->Table', 't1.id->t1',
+ 'Column->Table', 't1.bar->t1',
+ 'ForeignKeyConstraint->Table',
+ 'ForeignKeyConstraint()->t1',
+ 't1->MetaData(None)',
+ 'Table->MetaData', 'Column->Table',
+ 't2.id->t2', 't2->MetaData(None)']
+ )
+
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
