Column('value', String(50))
)
- The Table object constructs a unique instance of itself based on its
- name within the given MetaData object. Constructor
- arguments are as follows:
+ The :class:`.Table` object constructs a unique instance of itself based on its
+ name and optionl schema name within the given :class:`.MetaData` object.
+ Calling the :class:`.Table`
+ constructor with the same name and same :class:`.MetaData` argument
+ a second time will return the *same* :class:`.Table` object - in this way
+ the :class:`.Table` constructor acts as a registry function.
+
+ Constructor arguments are as follows:
:param name: The name of this table as represented in the database.
or Connection instance to be used for the table reflection. If
``None``, the underlying MetaData's bound connectable will be used.
+ :param extend_existing: When ``True``, indicates that if this Table is already
+ present in the given :class:`.MetaData`, apply further arguments within
+ the constructor to the existing :class:`.Table`.
+
+ If extend_existing or keep_existing are not set, an error is
+ raised if additional table modifiers are specified when
+ the given :class:`.Table` is already present in the :class:`.MetaData`.
+
:param implicit_returning: True by default - indicates that
RETURNING can be used by default to fetch newly inserted primary key
values, for backends which support this. Note that
:param info: A dictionary which defaults to ``{}``. A space to store
application specific data. This must be a dictionary.
+ :param keep_existing: When ``True``, indicates that if this Table
+ is already present in the given :class:`.MetaData`, ignore
+ further arguments within the constructor to the existing
+ :class:`.Table`, and return the :class:`.Table` object as
+ originally created. This is to allow a function that wishes
+ to define a new :class:`.Table` on first call, but on
+ subsequent calls will return the same :class:`.Table`,
+ without any of the declarations (particularly constraints)
+ being applied a second time. Also see extend_existing.
+
+ If extend_existing or keep_existing are not set, an error is
+ raised if additional table modifiers are specified when
+ the given :class:`.Table` is already present in the :class:`.MetaData`.
+
:param listeners: A list of tuples of the form ``(<eventname>, <fn>)``
which will be passed to :func:`.event.listen` upon construction.
This alternate hook to :func:`.event.listen` allows the establishment
the :meth:`.events.column_reflect` event::
def listen_for_reflect(table, column_info):
+ "handle the column reflection event"
# ...
t = Table(
])
:param mustexist: When ``True``, indicates that this Table must already
- be present in the given :class:`.MetaData`` collection.
+ be present in the given :class:`.MetaData`` collection, else
+ an exception is raised.
:param prefixes:
A list of strings to insert after CREATE in the CREATE TABLE
:param schema: The *schema name* for this table, which is required if
the table resides in a schema other than the default selected schema
for the engine's database connection. Defaults to ``None``.
-
- :param useexisting: When ``True``, indicates that if this Table is already
- present in the given :class:`.MetaData`, apply further arguments within
- the constructor to the existing :class:`.Table`. If this flag is not
- set, an error is raised when the parameters of an existing
- :class:`.Table` are overwritten.
+
+ :param useexisting: Deprecated. Use extend_existing.
"""
raise TypeError("Table() takes at least two arguments")
schema = kw.get('schema', None)
- useexisting = kw.pop('useexisting', False)
+ keep_existing = kw.pop('keep_existing', False)
+ extend_existing = kw.pop('extend_existing', False)
+ if 'useexisting' in kw:
+ util.warn_deprecated("useexisting is deprecated. Use extend_existing.")
+ if extend_existing:
+ raise exc.ArgumentError("useexisting is synonymous "
+ "with extend_existing.")
+ extend_existing = kw.pop('useexisting', False)
+
+ if keep_existing and extend_existing:
+ raise exc.ArgumentError("keep_existing and extend_existing "
+ "are mutually exclusive.")
+
mustexist = kw.pop('mustexist', False)
key = _get_table_key(name, schema)
if key in metadata.tables:
- if not useexisting and bool(args):
+ if not keep_existing and not extend_existing and bool(args):
raise exc.InvalidRequestError(
"Table '%s' is already defined for this MetaData "
- "instance. Specify 'useexisting=True' to redefine "
- "options and columns on an existing Table object." % key)
+ "instance. Specify 'extend_existing=True' "
+ "to redefine "
+ "options and columns on an "
+ "existing Table object." % key)
table = metadata.tables[key]
- table._init_existing(*args, **kw)
+ if extend_existing:
+ table._init_existing(*args, **kw)
return table
else:
if mustexist:
from sqlalchemy import types as sql_types
from sqlalchemy import schema, events, event
from sqlalchemy.engine.reflection import Inspector
-from sqlalchemy import MetaData
+from sqlalchemy import MetaData, Integer
from test.lib.schema import Table, Column
import sqlalchemy as sa
from test.lib import ComparesTables, \
meta3 = MetaData(testing.db)
foo = Table('foo', meta3, autoload=True)
foo = Table('foo', meta3, include_columns=['b', 'f', 'e'],
- useexisting=True)
+ extend_existing=True)
eq_([c.name for c in foo.c], ['b', 'e', 'f'])
for c in ('b', 'f', 'e'):
assert c in foo.c
finally:
meta.drop_all()
- @testing.exclude('mysql', '<', (4, 1, 1), 'innodb funkiness')
- def test_use_existing(self):
- meta = MetaData(testing.db)
- users = Table('users', meta,
- Column('id', sa.Integer, primary_key=True),
- Column('name', sa.String(30)),
- test_needs_fk=True)
- addresses = Table(
- 'addresses',
- meta,
- Column('id', sa.Integer, primary_key=True),
- Column('user_id', sa.Integer, sa.ForeignKey('users.id')),
- Column('data', sa.String(100)),
- test_needs_fk=True,
- )
- meta.create_all()
- try:
- meta2 = MetaData(testing.db)
- addresses = Table('addresses', meta2, Column('data',
- sa.Unicode), autoload=True)
- try:
- users = Table('users', meta2, Column('name',
- sa.Unicode), autoload=True)
- assert False
- except sa.exc.InvalidRequestError, err:
- assert str(err) \
- == "Table 'users' is already defined for this "\
- "MetaData instance. Specify 'useexisting=True' "\
- "to redefine options and columns on an existing "\
- "Table object."
- users = Table('users', meta2, Column('name', sa.Unicode),
- autoload=True, useexisting=True)
- assert isinstance(users.c.name.type, sa.Unicode)
- assert not users.quote
- users = Table('users', meta2, quote=True, autoload=True,
- useexisting=True)
- assert users.quote
- finally:
- meta.drop_all()
-
def test_pks_not_uniques(self):
"""test that primary key reflection not tripped up by unique
indexes"""
from sqlalchemy import Integer, String, UniqueConstraint, \
CheckConstraint, ForeignKey, MetaData, Sequence, \
ForeignKeyConstraint, ColumnDefault, Index, event,\
- events
+ events, Unicode
from test.lib.schema import Table, Column
from sqlalchemy import schema, exc
import sqlalchemy as tsa
assert_raises_message(
tsa.exc.InvalidRequestError,
"Table 'table1' is already defined for this "\
- "MetaData instance. Specify 'useexisting=True' "\
+ "MetaData instance. Specify 'extend_existing=True' "\
"to redefine options and columns on an existing "\
"Table object.",
go
t2 = Table('a', m2,
Column('id',Integer,primary_key=True),
Column('x', Integer, s2),
- useexisting=True)
+ extend_existing=True)
assert m2._sequences['x_seq'] is t2.c.x.default
assert m2._sequences['x_seq'] is s2
s2 = Sequence('x_seq')
t2 = Table('a', m1,
Column('x', Integer, s2),
- useexisting=True
+ extend_existing=True
)
assert t.c.x.default is s2
assert m1._sequences['x_seq'] is s2
m2 = pickle.loads(pickle.dumps(m1))
- t2 = Table('a', m2, useexisting=True)
+ t2 = Table('a', m2, extend_existing=True)
eq_(m2._sequences, {'x_seq':t2.c.x.default})
m2 = pickle.loads(pickle.dumps(m1))
t2 = Table('a', m2, schema='y',
- useexisting=True)
+ extend_existing=True)
eq_(m2._schemas, m1._schemas)
assign
)
+class UseExistingTest(fixtures.TablesTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('users', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('name', String(30)))
+
+ def _useexisting_fixture(self):
+ meta2 = MetaData(testing.db)
+ Table('users', meta2, autoload=True)
+ return meta2
+
+ def _notexisting_fixture(self):
+ return MetaData(testing.db)
+
+ def test_exception_no_flags(self):
+ meta2 = self._useexisting_fixture()
+ def go():
+ users = Table('users', meta2, Column('name',
+ Unicode), autoload=True)
+ assert_raises_message(
+ exc.InvalidRequestError,
+ "Table 'users' is already defined for this "\
+ "MetaData instance.",
+ go
+ )
+
+ @testing.uses_deprecated
+ def test_deprecated_useexisting(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ autoload=True, useexisting=True)
+ assert isinstance(users.c.name.type, Unicode)
+ assert not users.quote
+ users = Table('users', meta2, quote=True, autoload=True,
+ useexisting=True)
+ assert users.quote
+
+ def test_keep_plus_existing_raises(self):
+ meta2 = self._useexisting_fixture()
+ assert_raises(
+ exc.ArgumentError,
+ Table, 'users', meta2, keep_existing=True,
+ extend_existing=True
+ )
+
+ @testing.uses_deprecated
+ def test_existing_plus_useexisting_raises(self):
+ meta2 = self._useexisting_fixture()
+ assert_raises(
+ exc.ArgumentError,
+ Table, 'users', meta2, useexisting=True,
+ extend_existing=True
+ )
+
+ def test_keep_existing_no_dupe_constraints(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2,
+ Column('id', Integer),
+ Column('name', Unicode),
+ UniqueConstraint('name'),
+ keep_existing=True
+ )
+ assert 'name' in users.c
+ assert 'id' in users.c
+ eq_(len(users.constraints), 2)
+
+ u2 = Table('users', meta2,
+ Column('id', Integer),
+ Column('name', Unicode),
+ UniqueConstraint('name'),
+ keep_existing=True
+ )
+ eq_(len(u2.constraints), 2)
+
+ def test_extend_existing_dupes_constraints(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2,
+ Column('id', Integer),
+ Column('name', Unicode),
+ UniqueConstraint('name'),
+ extend_existing=True
+ )
+ assert 'name' in users.c
+ assert 'id' in users.c
+ eq_(len(users.constraints), 2)
+
+ u2 = Table('users', meta2,
+ Column('id', Integer),
+ Column('name', Unicode),
+ UniqueConstraint('name'),
+ extend_existing=True
+ )
+ # constraint got duped
+ eq_(len(u2.constraints), 3)
+
+ def test_keep_existing_coltype(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ autoload=True, keep_existing=True)
+ assert not isinstance(users.c.name.type, Unicode)
+
+ def test_keep_existing_quote(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, quote=True, autoload=True,
+ keep_existing=True)
+ assert not users.quote
+
+ def test_keep_existing_add_column(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ autoload=True,
+ keep_existing=True)
+ assert "foo" not in users.c
+
+ def test_keep_existing_coltype_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ autoload=True, keep_existing=True)
+ assert isinstance(users.c.name.type, Unicode)
+
+ def test_keep_existing_quote_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2, quote=True,
+ autoload=True,
+ keep_existing=True)
+ assert users.quote
+
+ def test_keep_existing_add_column_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ autoload=True,
+ keep_existing=True)
+ assert "foo" in users.c
+
+ def test_keep_existing_coltype_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ keep_existing=True)
+ assert not isinstance(users.c.name.type, Unicode)
+
+ def test_keep_existing_quote_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, quote=True,
+ keep_existing=True)
+ assert not users.quote
+
+ def test_keep_existing_add_column_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ keep_existing=True)
+ assert "foo" not in users.c
+
+ def test_extend_existing_coltype(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ autoload=True, extend_existing=True)
+ assert isinstance(users.c.name.type, Unicode)
+
+ def test_extend_existing_quote(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, quote=True, autoload=True,
+ extend_existing=True)
+ assert users.quote
+
+ def test_extend_existing_add_column(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ autoload=True,
+ extend_existing=True)
+ assert "foo" in users.c
+
+ def test_extend_existing_coltype_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ autoload=True, extend_existing=True)
+ assert isinstance(users.c.name.type, Unicode)
+
+ def test_extend_existing_quote_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2, quote=True,
+ autoload=True,
+ extend_existing=True)
+ assert users.quote
+
+ def test_extend_existing_add_column_no_orig(self):
+ meta2 = self._notexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ autoload=True,
+ extend_existing=True)
+ assert "foo" in users.c
+
+ def test_extend_existing_coltype_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, Column('name', Unicode),
+ extend_existing=True)
+ assert isinstance(users.c.name.type, Unicode)
+
+ def test_extend_existing_quote_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2, quote=True,
+ extend_existing=True)
+ assert users.quote
+
+ def test_extend_existing_add_column_no_reflection(self):
+ meta2 = self._useexisting_fixture()
+ users = Table('users', meta2,
+ Column('foo', Integer),
+ extend_existing=True)
+ assert "foo" in users.c
+
class ConstraintTest(fixtures.TestBase):
def _single_fixture(self):
m = MetaData()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
