0.7.4
=====
- examples
- - Fixed bug in history_meta.py example where
+ - [bug] Fixed bug in history_meta.py example where
the "unique" flag was not removed from a
single-table-inheritance subclass which
generates columns to put up onto the base.
- orm
- - Added missing comma to PASSIVE_RETURN_NEVER_SET
+ - [bug] Added missing comma to PASSIVE_RETURN_NEVER_SET
symbol [ticket:2304]
- - Cls.column.collate("some collation") now
+ - [bug] Cls.column.collate("some collation") now
works. [ticket:1776] Also in 0.6.9
- sql
- - Added accessor to types called "python_type",
+ - [feature] Added accessor to types called "python_type",
returns the rudimentary Python type object
for a particular TypeEngine instance, if known,
else raises NotImplementedError. [ticket:77]
+- schema
+ - [feature] Added new support for remote "schemas":
+ - MetaData() accepts "schema" and "quote_schema"
+ arguments, which will be applied to the same-named
+ arguments of a Table
+ or Sequence which leaves these at their default
+ of ``None``.
+ - Sequence accepts "quote_schema" argument
+ - tometadata() for Table will use the "schema"
+ of the incoming MetaData for the new Table
+ if the schema argument is explicitly "None"
+ - Added CreateSchema and DropSchema DDL
+ constructs - these accept just the string
+ name of a schema and a "quote" flag.
+ - When using default "schema" with MetaData,
+ ForeignKey will also assume the "default" schema
+ when locating remote table. This allows the "schema"
+ argument on MetaData to be applied to any
+ set of Table objects that otherwise don't have
+ a "schema".
+ - a "has_schema" method has been implemented
+ on dialect, but only works on Postgresql so far.
+ Courtesy Manlio Perillo, [ticket:1679]
+
0.7.3
=====
- general
:undoc-members:
:show-inheritance:
+.. autoclass:: CreateSchema
+ :members:
+ :undoc-members:
+ :show-inheritance:
+
+.. autoclass:: DropSchema
+ :members:
+ :undoc-members:
+ :show-inheritance:
+
def _get_default_schema_name(self, connection):
return connection.scalar("select current_schema()")
+ def has_schema(self, connection, schema):
+ cursor = connection.execute(
+ sql.text(
+ "select nspname from pg_namespace where lower(nspname)=:schema",
+ bindparams=[
+ sql.bindparam(
+ 'schema', unicode(schema.lower()),
+ type_=sqltypes.Unicode)]
+ )
+ )
+
+ return bool(cursor.first())
+
def has_table(self, connection, table_name, schema=None):
# seems like case gets folded in pg_class...
if schema is None:
raise TypeError("Table() takes at least two arguments")
schema = kw.get('schema', None)
+ if schema is None:
+ schema = metadata.schema
keep_existing = kw.pop('keep_existing', False)
extend_existing = kw.pop('extend_existing', False)
if 'useexisting' in kw:
super(Table, self).__init__(name)
self.metadata = metadata
self.schema = kwargs.pop('schema', None)
+ if self.schema is None:
+ self.schema = metadata.schema
+ self.quote_schema = kwargs.pop('quote_schema', metadata.quote_schema)
+ else:
+ self.quote_schema = kwargs.pop('quote_schema', None)
+
self.indexes = set()
self.constraints = set()
self._columns = expression.ColumnCollection()
self.implicit_returning = kwargs.pop('implicit_returning', True)
self.quote = kwargs.pop('quote', None)
- self.quote_schema = kwargs.pop('quote_schema', None)
if 'info' in kwargs:
self.info = kwargs.pop('info')
if 'listeners' in kwargs:
if schema is RETAIN_SCHEMA:
schema = self.schema
+ elif schema is None:
+ schema = metadata.schema
key = _get_table_key(self.name, schema)
if key in metadata.tables:
util.warn("Table '%s' already exists within the given "
def __init__(self, column, _constraint=None, use_alter=False, name=None,
onupdate=None, ondelete=None, deferrable=None,
+ schema=None,
initially=None, link_to_name=False):
"""
Construct a column-level FOREIGN KEY.
(defaults to the column name itself), unless ``link_to_name`` is
``True`` in which case the rendered name of the column is used.
+ Note that if the schema name is not included, and the underlying
+ :class:`.MetaData` has a "schema", that value will be used.
+ (new in 0.7.4)
+
:param name: Optional string. An in-database name for the key if
`constraint` is not provided.
# will never appear *within* any component of the FK.
(schema, tname, colname) = (None, None, None)
+ if schema is None and parenttable.metadata.schema is not None:
+ schema = parenttable.metadata.schema
+
if (len(m) == 1):
tname = m.pop()
else:
def __init__(self, name, start=None, increment=None, schema=None,
optional=False, quote=None, metadata=None,
+ quote_schema=None,
for_update=False):
"""Construct a :class:`.Sequence` object.
self.increment = increment
self.optional = optional
self.quote = quote
- self.schema = schema
+ if metadata is not None and schema is None and metadata.schema:
+ self.schema = schema = metadata.schema
+ self.quote_schema = metadata.quote_schema
+ else:
+ self.schema = schema
+ self.quote_schema = quote_schema
self.metadata = metadata
self._key = _get_table_key(name, schema)
if metadata:
__visit_name__ = 'metadata'
- def __init__(self, bind=None, reflect=False):
+ def __init__(self, bind=None, reflect=False, schema=None, quote_schema=None):
"""Create a new MetaData object.
:param bind:
For finer control over loaded tables, use the ``reflect`` method of
``MetaData``.
+ :param schema:
+ The default schema to use for the :class:`.Table`, :class:`.Sequence`, and other
+ objects associated with this :class:`.MetaData`.
+ Defaults to ``None``. New in 0.7.4.
+
+ :param quote_schema:
+ Sets the ``quote_schema`` flag for those :class:`.Table`, :class:`.Sequence`,
+ and other objects which make usage of the local ``schema`` name.
+ New in 0.7.4.
+
"""
self.tables = util.immutabledict()
+ self.schema = schema
+ self.quote_schema = quote_schema
self._schemas = set()
self._sequences = {}
self.bind = bind
if t.schema is not None])
def __getstate__(self):
- return {'tables': self.tables, 'schemas':self._schemas,
+ return {'tables': self.tables, 'schema':self.schema,
+ 'quote_schema':self.quote_schema,
+ 'schemas':self._schemas,
'sequences':self._sequences}
def __setstate__(self, state):
self.tables = state['tables']
+ self.schema = state['schema']
+ self.quote_schema = state['quote_schema']
self._bind = None
self._sequences = state['sequences']
self._schemas = state['schemas']
:param schema:
Optional, query and reflect tables from an alterate schema.
+ If None, the schema associated with this :class:`.MetaData`
+ is used, if any.
:param views:
If True, also reflect views.
reflect_opts['autoload_with'] = bind
conn = bind.contextual_connect()
+ if schema is None:
+ schema = self.schema
+
if schema is not None:
reflect_opts['schema'] = schema
"""
return False
+class CreateSchema(_CreateDropBase):
+ """Represent a CREATE SCHEMA statement.
+
+ New in 0.7.4.
+
+ The argument here is the string name of the schema.
+
+ """
+
+ __visit_name__ = "create_schema"
+
+ def __init__(self, name, quote=None, **kw):
+ """Create a new :class:`.CreateSchema` construct."""
+
+ self.quote = quote
+ super(CreateSchema, self).__init__(name, **kw)
+
+class DropSchema(_CreateDropBase):
+ """Represent a DROP SCHEMA statement.
+
+ The argument here is the string name of the schema.
+
+ New in 0.7.4.
+ """
+
+ __visit_name__ = "drop_schema"
+
+ def __init__(self, name, quote=None, cascade=False, **kw):
+ """Create a new :class:`.DropSchema` construct."""
+
+ self.quote = quote
+ self.cascade=cascade
+ super(DropSchema, self).__init__(name, **kw)
+
+
class CreateTable(_CreateDropBase):
"""Represent a CREATE TABLE statement."""
if colparams or not supports_default_values:
text += " (%s)" % ', '.join([preparer.format_column(c[0])
for c in colparams])
-
+
if self.returning or insert_stmt._returning:
self.returning = self.returning or insert_stmt._returning
returning_clause = self.returning_clause(
return self.sql_compiler.post_process_text(ddl.statement % context)
+ def visit_create_schema(self, create):
+ return "CREATE SCHEMA " + self.preparer.format_schema(create.element, create.quote)
+
+ def visit_drop_schema(self, drop):
+ text = "DROP SCHEMA " + self.preparer.format_schema(drop.element, drop.quote)
+ if drop.cascade:
+ text += " CASCADE"
+ return text
+
def visit_create_table(self, create):
table = create.element
preparer = self.dialect.identifier_preparer
"." + result
return result
+ def format_schema(self, name, quote):
+ """Prepare a quoted schema name."""
+
+ return self.quote(name, quote)
+
def format_column(self, column, use_table=False,
name=None, table_name=None):
"""Prepare a quoted column name."""
class SchemaTest(fixtures.TestBase):
- def test_iteration(self):
- metadata = MetaData()
- table1 = Table('table1', metadata, Column('col1', sa.Integer,
- primary_key=True), schema='someschema')
- table2 = Table('table2', metadata, Column('col1', sa.Integer,
- primary_key=True), Column('col2', sa.Integer,
- sa.ForeignKey('someschema.table1.col1')),
- schema='someschema')
-
- t1 = str(schema.CreateTable(table1).compile(bind=testing.db))
- t2 = str(schema.CreateTable(table2).compile(bind=testing.db))
- if testing.db.dialect.preparer(testing.db.dialect).omit_schema:
- assert t1.index("CREATE TABLE table1") > -1
- assert t2.index("CREATE TABLE table2") > -1
- else:
- assert t1.index("CREATE TABLE someschema.table1") > -1
- assert t2.index("CREATE TABLE someschema.table2") > -1
+ @testing.requires.schemas
+ @testing.fails_on_everything_except("postgresql", "unimplemented feature")
+ def test_has_schema(self):
+ eq_(testing.db.dialect.has_schema(testing.db, 'test_schema'), True)
+ eq_(testing.db.dialect.has_schema(testing.db, 'sa_fake_schema_123'), False)
@testing.crashes('firebird', 'No schema support')
@testing.fails_on('sqlite', 'FIXME: unknown')
finally:
metadata.drop_all()
+ @testing.crashes('firebird', 'No schema support')
+ # fixme: revisit these below.
+ @testing.fails_on('access', 'FIXME: unknown')
+ @testing.fails_on('sybase', 'FIXME: unknown')
+ def test_explicit_default_schema_metadata(self):
+ engine = testing.db
+
+ if testing.against('sqlite'):
+ # Works for CREATE TABLE main.foo, SELECT FROM main.foo, etc.,
+ # but fails on:
+ # FOREIGN KEY(col2) REFERENCES main.table1 (col1)
+ schema = 'main'
+ else:
+ schema = engine.dialect.default_schema_name
+
+ assert bool(schema)
+
+ metadata = MetaData(engine, schema=schema)
+ table1 = Table('table1', metadata,
+ Column('col1', sa.Integer, primary_key=True),
+ test_needs_fk=True)
+ table2 = Table('table2', metadata,
+ Column('col1', sa.Integer, primary_key=True),
+ Column('col2', sa.Integer,
+ sa.ForeignKey('table1.col1')),
+ test_needs_fk=True)
+ try:
+ metadata.create_all()
+ metadata.create_all(checkfirst=True)
+ assert len(metadata.tables) == 2
+ metadata.clear()
+
+ table1 = Table('table1', metadata, autoload=True)
+ table2 = Table('table2', metadata, autoload=True)
+ assert len(metadata.tables) == 2
+ finally:
+ metadata.drop_all()
+
+ @testing.requires.schemas
+ @testing.provide_metadata
+ def test_metadata_reflect_schema(self):
+ metadata = self.metadata
+ createTables(metadata, "test_schema")
+ metadata.create_all()
+ m2 = MetaData(schema="test_schema", bind=testing.db)
+ m2.reflect()
+ eq_(m2.tables.keys(),
+ [u'test_schema.users', u'test_schema.email_addresses']
+ )
class HasSequenceTest(fixtures.TestBase):
eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
False)
+
+
# Tests related to engine.reflection
eq_(str(table_c.join(table2_c).onclause),
'someschema.mytable.myid = someschema.othertable.myid')
+ def test_tometadata_with_default_schema(self):
+ meta = MetaData()
+
+ table = Table('mytable', meta,
+ Column('myid', Integer, primary_key=True),
+ Column('name', String(40), nullable=True),
+ Column('description', String(30),
+ CheckConstraint("description='hi'")),
+ UniqueConstraint('name'),
+ test_needs_fk=True,
+ schema='myschema',
+ )
+
+ table2 = Table('othertable', meta,
+ Column('id', Integer, primary_key=True),
+ Column('myid', Integer, ForeignKey('myschema.mytable.myid')),
+ test_needs_fk=True,
+ schema='myschema',
+ )
+
+ meta2 = MetaData()
+ table_c = table.tometadata(meta2)
+ table2_c = table2.tometadata(meta2)
+
+ eq_(str(table_c.join(table2_c).onclause), str(table_c.c.myid
+ == table2_c.c.myid))
+ eq_(str(table_c.join(table2_c).onclause),
+ 'myschema.mytable.myid = myschema.othertable.myid')
+
+
def test_tometadata_kwargs(self):
meta = MetaData()
# d'oh!
assert table_c is table_d
- def test_tometadata_default_schema(self):
- meta = MetaData()
-
- table = Table('mytable', meta,
- Column('myid', Integer, primary_key=True),
- Column('name', String(40), nullable=True),
- Column('description', String(30),
- CheckConstraint("description='hi'")),
- UniqueConstraint('name'),
- test_needs_fk=True,
- schema='myschema',
- )
-
- table2 = Table('othertable', meta,
- Column('id', Integer, primary_key=True),
- Column('myid', Integer, ForeignKey('myschema.mytable.myid')),
- test_needs_fk=True,
- schema='myschema',
- )
-
- meta2 = MetaData()
- table_c = table.tometadata(meta2)
- table2_c = table2.tometadata(meta2)
-
- eq_(str(table_c.join(table2_c).onclause), str(table_c.c.myid
- == table2_c.c.myid))
- eq_(str(table_c.join(table2_c).onclause),
- 'myschema.mytable.myid = myschema.othertable.myid')
+ def test_metadata_schema_arg(self):
+ m1 = MetaData(schema='sch1')
+ m2 = MetaData(schema='sch1', quote_schema=True)
+ m3 = MetaData(schema='sch1', quote_schema=False)
+ m4 = MetaData()
+
+ for i, (name, metadata, schema, quote_schema, exp_schema, exp_quote_schema) in enumerate([
+ ('t1', m1, None, None, 'sch1', None),
+ ('t2', m1, 'sch2', None, 'sch2', None),
+ ('t3', m1, 'sch2', True, 'sch2', True),
+ ('t4', m1, 'sch1', None, 'sch1', None),
+ ('t1', m2, None, None, 'sch1', True),
+ ('t2', m2, 'sch2', None, 'sch2', None),
+ ('t3', m2, 'sch2', True, 'sch2', True),
+ ('t4', m2, 'sch1', None, 'sch1', None),
+ ('t1', m3, None, None, 'sch1', False),
+ ('t2', m3, 'sch2', None, 'sch2', None),
+ ('t3', m3, 'sch2', True, 'sch2', True),
+ ('t4', m3, 'sch1', None, 'sch1', None),
+ ('t1', m4, None, None, None, None),
+ ('t2', m4, 'sch2', None, 'sch2', None),
+ ('t3', m4, 'sch2', True, 'sch2', True),
+ ('t4', m4, 'sch1', None, 'sch1', None),
+ ]):
+ kw = {}
+ if schema is not None:
+ kw['schema'] = schema
+ if quote_schema is not None:
+ kw['quote_schema'] = quote_schema
+ t = Table(name, metadata, **kw)
+ eq_(t.schema, exp_schema, "test %d, table schema" % i)
+ eq_(t.quote_schema, exp_quote_schema, "test %d, table quote_schema" % i)
+ seq = Sequence(name, metadata=metadata, **kw)
+ eq_(seq.schema, exp_schema, "test %d, seq schema" % i)
+ eq_(seq.quote_schema, exp_quote_schema, "test %d, seq quote_schema" % i)
def test_manual_dependencies(self):
meta = MetaData()
[d, b, a, c, e]
)
+ def test_tometadata_default_schema_metadata(self):
+ meta = MetaData(schema='myschema')
+
+ table = Table('mytable', meta,
+ Column('myid', Integer, primary_key=True),
+ Column('name', String(40), nullable=True),
+ Column('description', String(30), CheckConstraint("description='hi'")),
+ UniqueConstraint('name'),
+ test_needs_fk=True
+ )
+
+ table2 = Table('othertable', meta,
+ Column('id', Integer, primary_key=True),
+ Column('myid', Integer, ForeignKey('myschema.mytable.myid')),
+ test_needs_fk=True
+ )
+
+ meta2 = MetaData(schema='someschema')
+ table_c = table.tometadata(meta2, schema=None)
+ table2_c = table2.tometadata(meta2, schema=None)
+
+ eq_(str(table_c.join(table2_c).onclause),
+ str(table_c.c.myid == table2_c.c.myid))
+ eq_(str(table_c.join(table2_c).onclause),
+ "someschema.mytable.myid = someschema.othertable.myid")
+
def test_tometadata_strip_schema(self):
meta = MetaData()
assign
)
+class SchemaTest(fixtures.TestBase, AssertsCompiledSQL):
+
+ def test_default_schema_metadata_fk(self):
+ m = MetaData(schema="foo")
+ t1 = Table('t1', m, Column('x', Integer))
+ t2 = Table('t2', m, Column('x', Integer, ForeignKey('t1.x')))
+ assert t2.c.x.references(t1.c.x)
+
+ def test_ad_hoc_schema_equiv_fk(self):
+ m = MetaData()
+ t1 = Table('t1', m, Column('x', Integer), schema="foo")
+ t2 = Table('t2', m, Column('x', Integer, ForeignKey('t1.x')), schema="foo")
+ assert_raises(
+ exc.NoReferencedTableError,
+ lambda: t2.c.x.references(t1.c.x)
+ )
+
+ def test_default_schema_metadata_fk_alt_remote(self):
+ m = MetaData(schema="foo")
+ t1 = Table('t1', m, Column('x', Integer))
+ t2 = Table('t2', m, Column('x', Integer, ForeignKey('t1.x')),
+ schema="bar")
+ assert t2.c.x.references(t1.c.x)
+
+ def test_default_schema_metadata_fk_alt_local_raises(self):
+ m = MetaData(schema="foo")
+ t1 = Table('t1', m, Column('x', Integer), schema="bar")
+ t2 = Table('t2', m, Column('x', Integer, ForeignKey('t1.x')))
+ assert_raises(
+ exc.NoReferencedTableError,
+ lambda: t2.c.x.references(t1.c.x)
+ )
+
+ def test_default_schema_metadata_fk_alt_local(self):
+ m = MetaData(schema="foo")
+ t1 = Table('t1', m, Column('x', Integer), schema="bar")
+ t2 = Table('t2', m, Column('x', Integer, ForeignKey('bar.t1.x')))
+ assert t2.c.x.references(t1.c.x)
+
+ def test_create_drop_schema(self):
+
+ self.assert_compile(
+ schema.CreateSchema("sa_schema"),
+ "CREATE SCHEMA sa_schema"
+ )
+ self.assert_compile(
+ schema.DropSchema("sa_schema"),
+ "DROP SCHEMA sa_schema"
+ )
+ self.assert_compile(
+ schema.DropSchema("sa_schema", cascade=True),
+ "DROP SCHEMA sa_schema CASCADE"
+ )
+
+ def test_iteration(self):
+ metadata = MetaData()
+ table1 = Table('table1', metadata, Column('col1', Integer,
+ primary_key=True), schema='someschema')
+ table2 = Table('table2', metadata, Column('col1', Integer,
+ primary_key=True), Column('col2', Integer,
+ ForeignKey('someschema.table1.col1')),
+ schema='someschema')
+
+ t1 = str(schema.CreateTable(table1).compile(bind=testing.db))
+ t2 = str(schema.CreateTable(table2).compile(bind=testing.db))
+ if testing.db.dialect.preparer(testing.db.dialect).omit_schema:
+ assert t1.index("CREATE TABLE table1") > -1
+ assert t2.index("CREATE TABLE table2") > -1
+ else:
+ assert t1.index("CREATE TABLE someschema.table1") > -1
+ assert t2.index("CREATE TABLE someschema.table2") > -1
+
+
class UseExistingTest(fixtures.TablesTest):
@classmethod
def define_tables(cls, metadata):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
