test user has DBA privs, and all objects can be created /dropped.
- added ORDER BY to oracle column listing
- Oracle all_tables always limits to current user if schema not given.
- views reflect - added documentation + a unit test for this.
- Table(autoload) with no bind produces an error message specific to
the fact that autoload_with should be the first option to try.
a wide variety of schema information, with room for expansion,
including table names, column names, view definitions, sequences,
indexes, etc.
-
+
+ - Views are now reflectable as ordinary Table objects. The same
+ Table constructor is used, with the caveat that "effective"
+ primary and foreign key constraints aren't part of the reflection
+ results; these have to be specified explicitly if desired.
+
- The existing autoload=True system now uses Inspector underneath
so that each dialect need only return "raw" data about tables
and other objects - Inspector is the single place that information
The test suite will be creating and dropping many tables and other DDL, and
preexisting tables will interfere with the tests.
-Several tests require alternate schemas to be present. This requirement
+Several tests require alternate schemas to be present. This requirement
applies to all backends except SQLite and Firebird. These schemas are:
test_schema
Additional steps specific to individual databases are as follows:
-Oracle: to run the test.dialect.test_oracle:MultiSchemaTest suite of tests,
-the database owner should be named "scott" (this will be fixed),
-and an additional "owner" named "ed" is required:
-
-1. create a user 'ed' in the oracle database.
-2. in 'ed', issue the following statements:
- create table parent(id integer primary key, data varchar2(50));
- create table child(id integer primary key, data varchar2(50),
- parent_id integer references parent(id));
- create synonym ptable for parent;
- create synonym ctable for child;
- grant all on parent to scott; (or to whoever you run the oracle tests as)
- grant all on child to scott; (same)
- grant all on ptable to scott;
- grant all on ctable to scott;
-
-MSSQL: Tests that involve multiple connections require Snapshot Isolation
-ability implented on the test database in order to prevent deadlocks that will
-occur with record locking isolation. This feature is only available with
-MSSQL 2005 and greater. You must enable snapshot isolation at the database level
-and set the default cursor isolation with two SQL commands:
-
- ALTER DATABASE MyDatabase
- SET ALLOW_SNAPSHOT_ISOLATION ON
+ ORACLE: the test_schema and test_schema_2 schemas are created as
+ users, as the "owner" in Oracle is considered like a "schema" in
+ SQLAlchemy.
- ALTER DATABASE MyDatabase
- SET READ_COMMITTED_SNAPSHOT ON
+ The primary database user needs to be able to create and drop tables,
+ synonyms, and constraints in these schemas. Unfortunately, many hours of
+ googling and experimentation cannot find a GRANT option that allows the
+ primary user the "REFERENCES" role in a remote schema for tables not yet
+ defined (REFERENCES is per-table) - the only thing that works is to put
+ the user in the "DBA" role:
+
+ grant dba to scott;
+
+ Any ideas on what specific privileges within "DBA" allow an open-ended
+ REFERENCES grant would be appreciated, or if in fact "DBA" has some kind
+ of "magic" flag not accessible otherwise. So, running SQLA tests on oracle
+ requires access to a completely open Oracle database - Oracle XE is
+ obviously a terrific choice since its just a local engine. As always,
+ leaving the schemas out means those few dozen tests will fail and is
+ otherwise harmless.
+
+
+ MSSQL: Tests that involve multiple connections require Snapshot Isolation
+ ability implented on the test database in order to prevent deadlocks that
+ will occur with record locking isolation. This feature is only available
+ with MSSQL 2005 and greater. You must enable snapshot isolation at the
+ database level and set the default cursor isolation with two SQL commands:
+
+ ALTER DATABASE MyDatabase SET ALLOW_SNAPSHOT_ISOLATION ON
+
+ ALTER DATABASE MyDatabase SET READ_COMMITTED_SNAPSHOT ON
-MSSQL+zxJDBC: Trying to run the unit tests on Windows against SQL Server
-requires using a test.cfg configuration file as the cmd.exe shell won't properly
-pass the URL arguments into the nose test runner.
+ MSSQL+zxJDBC: Trying to run the unit tests on Windows against SQL Server
+ requires using a test.cfg configuration file as the cmd.exe shell won't
+ properly pass the URL arguments into the nose test runner.
If you'll be running the tests frequently, database aliases can save a lot of
typing. The --dbs option lists the built-in aliases and their matching URLs:
... Column('mydata', Unicode(50)), # override reflected 'mydata' to be Unicode
... autoload=True)
+Reflecting Views
+~~~~~~~~~~~~~~~~
+
+The reflection system can also reflect views. Basic usage is the same as that of a table::
+
+ my_view = Table("some_view", metadata, autoload=True)
+
+Above, ``my_view`` is a ``Table`` object with ``Column`` objects representing the names and types
+of each column within the view "some_view".
+
+Usually, it's desired to have at least a primary key constraint when reflecting a view, if not
+foreign keys as well. View reflection doesn't extrapolate these constraints.
+
+Use the "override" technique for this, specifying explicitly those columns
+which are part of the primary key or have foreign key constraints::
+
+ my_view = Table("some_view", metadata,
+ Column("view_id", Integer, primary_key=True),
+ Column("related_thing", Integer, ForeignKey("othertable.thing_id")),
+ autoload=True
+ )
+
Reflecting All Tables at Once
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def table_names(self, connection, schema):
# note that table_names() isnt loading DBLINKed or synonym'ed tables
if schema is None:
- cursor = connection.execute(
- "SELECT table_name FROM all_tables "
- "WHERE nvl(tablespace_name, 'no tablespace') NOT IN ('SYSTEM', 'SYSAUX') "
- "AND IOT_NAME IS NULL")
- else:
- s = sql.text(
- "SELECT table_name FROM all_tables "
- "WHERE nvl(tablespace_name, 'no tablespace') NOT IN ('SYSTEM', 'SYSAUX') "
- "AND OWNER = :owner "
- "AND IOT_NAME IS NULL")
- cursor = connection.execute(s, owner=self.denormalize_name(schema))
+ schema = self.default_schema_name
+ s = sql.text(
+ "SELECT table_name FROM all_tables "
+ "WHERE nvl(tablespace_name, 'no tablespace') NOT IN ('SYSTEM', 'SYSAUX') "
+ "AND OWNER = :owner "
+ "AND IOT_NAME IS NULL")
+ cursor = connection.execute(s, owner=self.denormalize_name(schema))
return [self.normalize_name(row[0]) for row in cursor]
def _resolve_synonym(self, connection, desired_owner=None, desired_synonym=None, desired_table=None):
c = connection.execute(sql.text(
"SELECT column_name, data_type, data_length, data_precision, data_scale, "
"nullable, data_default FROM ALL_TAB_COLUMNS%(dblink)s "
- "WHERE table_name = :table_name AND owner = :owner" % {'dblink': dblink}),
+ "WHERE table_name = :table_name AND owner = :owner "
+ "ORDER BY column_id" % {'dblink': dblink}),
table_name=table_name, owner=schema)
for row in c:
if autoload_with:
autoload_with.reflecttable(self, include_columns=include_columns)
else:
- _bind_or_error(metadata).reflecttable(self, include_columns=include_columns)
+ _bind_or_error(metadata, msg="No engine is bound to this Table's MetaData. "
+ "Pass an engine to the Table via "
+ "autoload_with=<someengine>, "
+ "or associate the MetaData with an engine via "
+ "metadata.bind=<someengine>").\
+ reflecttable(self, include_columns=include_columns)
# initialize all the column, etc. objects. done after reflection to
# allow user-overrides
super(DropConstraint, self).__init__(element, **kw)
element._create_rule = lambda compiler: False
-def _bind_or_error(schemaitem):
+def _bind_or_error(schemaitem, msg=None):
bind = schemaitem.bind
if not bind:
name = schemaitem.__class__.__name__
bindable = "the %s's .bind" % name
else:
bindable = "this %s's .metadata.bind" % name
-
- msg = ('The %s is not bound to an Engine or Connection. '
- 'Execution can not proceed without a database to execute '
- 'against. Either execute with an explicit connection or '
- 'assign %s to enable implicit execution.') % (item, bindable)
+
+ if msg is None:
+ msg = ('The %s is not bound to an Engine or Connection. '
+ 'Execution can not proceed without a database to execute '
+ 'against. Either execute with an explicit connection or '
+ 'assign %s to enable implicit execution.') % (item, bindable)
raise exc.UnboundExecutionError(msg)
return bind
"address_types.id")
class MultiSchemaTest(TestBase, AssertsCompiledSQL):
- """instructions:
-
- 1. create a user 'ed' in the oracle database.
- 2. in 'ed', issue the following statements:
- create table parent(id integer primary key, data varchar2(50));
- create table child(id integer primary key, data varchar2(50), parent_id integer references parent(id));
- create synonym ptable for parent;
- create synonym ctable for child;
- grant all on parent to scott; (or to whoever you run the oracle tests as)
- grant all on child to scott; (same)
- grant all on ptable to scott;
- grant all on ctable to scott;
-
- """
-
__only_on__ = 'oracle'
-
+
+ @classmethod
+ def setup_class(cls):
+ # currently assuming full DBA privs for the user.
+ # don't really know how else to go here unless
+ # we connect as the other user.
+
+ for stmt in """
+create table test_schema.parent(
+ id integer primary key,
+ data varchar2(50)
+);
+
+create table test_schema.child(
+ id integer primary key,
+ data varchar2(50),
+ parent_id integer references test_schema.parent(id)
+);
+
+create synonym test_schema.ptable for test_schema.parent;
+create synonym test_schema.ctable for test_schema.child;
+
+-- can't make a ref from local schema to the remote schema's table without this,
+-- *and* cant give yourself a grant ! so we give it to public. ideas welcome.
+grant references on test_schema.parent to public;
+grant references on test_schema.child to public;
+""".split(";"):
+ if stmt.strip():
+ testing.db.execute(stmt)
+
+ @classmethod
+ def teardown_class(cls):
+ for stmt in """
+drop table test_schema.child;
+drop table test_schema.parent;
+drop synonym test_schema.ctable;
+drop synonym test_schema.ptable;
+""".split(";"):
+ if stmt.strip():
+ testing.db.execute(stmt)
+
def test_create_same_names_explicit_schema(self):
schema = testing.db.dialect.default_schema_name
meta = MetaData(testing.db)
)
child = Table('child', meta,
Column('cid', Integer, primary_key=True),
- Column('pid', Integer, ForeignKey('scott.parent.pid')),
+ Column('pid', Integer, ForeignKey('%s.parent.pid' % schema)),
schema=schema
)
meta.create_all()
def test_reflect_alt_owner_explicit(self):
meta = MetaData(testing.db)
- parent = Table('parent', meta, autoload=True, schema='ed')
- child = Table('child', meta, autoload=True, schema='ed')
+ parent = Table('parent', meta, autoload=True, schema='test_schema')
+ child = Table('child', meta, autoload=True, schema='test_schema')
- self.assert_compile(parent.join(child), "ed.parent JOIN ed.child ON ed.parent.id = ed.child.parent_id")
+ self.assert_compile(parent.join(child), "test_schema.parent JOIN test_schema.child ON test_schema.parent.id = test_schema.child.parent_id")
select([parent, child]).select_from(parent.join(child)).execute().fetchall()
def test_reflect_local_to_remote(self):
- testing.db.execute("CREATE TABLE localtable (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES ed.parent(id))")
+ testing.db.execute("CREATE TABLE localtable "
+ "(id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES"
+ " test_schema.parent(id))")
try:
meta = MetaData(testing.db)
lcl = Table('localtable', meta, autoload=True)
- parent = meta.tables['ed.parent']
- self.assert_compile(parent.join(lcl), "ed.parent JOIN localtable ON ed.parent.id = localtable.parent_id")
+ parent = meta.tables['test_schema.parent']
+ self.assert_compile(parent.join(lcl), "test_schema.parent JOIN localtable ON test_schema.parent.id = localtable.parent_id")
select([parent, lcl]).select_from(parent.join(lcl)).execute().fetchall()
finally:
testing.db.execute("DROP TABLE localtable")
def test_reflect_alt_owner_implicit(self):
meta = MetaData(testing.db)
- parent = Table('parent', meta, autoload=True, schema='ed')
- child = Table('child', meta, autoload=True, schema='ed')
+ parent = Table('parent', meta, autoload=True, schema='test_schema')
+ child = Table('child', meta, autoload=True, schema='test_schema')
- self.assert_compile(parent.join(child), "ed.parent JOIN ed.child ON ed.parent.id = ed.child.parent_id")
+ self.assert_compile(parent.join(child), "test_schema.parent JOIN test_schema.child ON test_schema.parent.id = test_schema.child.parent_id")
select([parent, child]).select_from(parent.join(child)).execute().fetchall()
def test_reflect_alt_owner_synonyms(self):
- testing.db.execute("CREATE TABLE localtable (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES ed.ptable(id))")
+ testing.db.execute("CREATE TABLE localtable "
+ "(id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES"
+ " test_schema.ptable(id))")
try:
meta = MetaData(testing.db)
lcl = Table('localtable', meta, autoload=True, oracle_resolve_synonyms=True)
- parent = meta.tables['ed.ptable']
- self.assert_compile(parent.join(lcl), "ed.ptable JOIN localtable ON ed.ptable.id = localtable.parent_id")
+ parent = meta.tables['test_schema.ptable']
+ self.assert_compile(parent.join(lcl), "test_schema.ptable JOIN localtable ON test_schema.ptable.id = localtable.parent_id")
select([parent, lcl]).select_from(parent.join(lcl)).execute().fetchall()
finally:
testing.db.execute("DROP TABLE localtable")
def test_reflect_remote_synonyms(self):
meta = MetaData(testing.db)
- parent = Table('ptable', meta, autoload=True, schema='ed', oracle_resolve_synonyms=True)
- child = Table('ctable', meta, autoload=True, schema='ed', oracle_resolve_synonyms=True)
- self.assert_compile(parent.join(child), "ed.ptable JOIN ed.ctable ON ed.ptable.id = ed.ctable.parent_id")
+ parent = Table('ptable', meta, autoload=True, schema='test_schema', oracle_resolve_synonyms=True)
+ child = Table('ctable', meta, autoload=True, schema='test_schema', oracle_resolve_synonyms=True)
+ self.assert_compile(parent.join(child), "test_schema.ptable JOIN test_schema.ctable ON test_schema.ptable.id = test_schema.ctable.parent_id")
select([parent, child]).select_from(parent.join(child)).execute().fetchall()
finally:
m1.drop_all()
+ def test_views(self):
+ meta = MetaData(testing.db)
+ users, addresses = createTables(meta, None)
+ meta.create_all()
+ createViews(meta.bind, None)
+ try:
+ m2 = MetaData(testing.db)
+ users_v = Table("users_v", m2, autoload=True)
+ addresses_v = Table("email_addresses_v", m2, autoload=True)
+
+ for c1, c2 in zip(users.c, users_v.c):
+ eq_(c1.name, c2.name)
+ assert c1.type._compare_type_affinity(c2.type)
+
+ for c1, c2 in zip(addresses.c, addresses_v.c):
+ eq_(c1.name, c2.name)
+ assert c1.type._compare_type_affinity(c2.type)
+
+ finally:
+ dropViews(meta.bind, None)
+ meta.drop_all()
+
class CreateDropTest(TestBase):
@classmethod
def setup_class(cls):
@testing.requires.schemas
@testing.requires.sequences
- @testing.fails_on('oracle', "the 'scott' schema is the same as the default schema")
def test_has_sequence_schema(self):
- test_schema = get_schema()
+ test_schema = 'test_schema'
s1 = sa.Sequence('user_id_seq', schema=test_schema)
s2 = sa.Sequence('user_id_seq')
testing.db.execute(schema.CreateSequence(s1))
# Tests related to engine.reflection
-def get_schema():
- # TODO: how to get multiple schemas in use on Oracle XE ?
- if testing.against('oracle'):
- return 'scott'
- return 'test_schema'
def createTables(meta, schema=None):
if schema:
def test_get_schema_names(self):
insp = Inspector(testing.db)
- self.assert_(get_schema() in insp.get_schema_names())
+ self.assert_('test_schema' in insp.get_schema_names())
def test_get_default_schema_name(self):
insp = Inspector(testing.db)
@testing.requires.schemas
def test_get_table_names_with_schema(self):
- self._test_get_table_names(get_schema())
+ self._test_get_table_names('test_schema')
def test_get_view_names(self):
self._test_get_table_names(table_type='view')
@testing.requires.schemas
def test_get_view_names_with_schema(self):
- self._test_get_table_names(get_schema(), table_type='view')
+ self._test_get_table_names('test_schema', table_type='view')
def _test_get_columns(self, schema=None, table_type='table'):
meta = MetaData(testing.db)
@testing.requires.schemas
def test_get_columns_with_schema(self):
- self._test_get_columns(schema=get_schema())
+ self._test_get_columns(schema='test_schema')
def test_get_view_columns(self):
self._test_get_columns(table_type='view')
@testing.requires.schemas
def test_get_view_columns_with_schema(self):
- self._test_get_columns(schema=get_schema(), table_type='view')
+ self._test_get_columns(schema='test_schema', table_type='view')
def _test_get_primary_keys(self, schema=None):
meta = MetaData(testing.db)
@testing.fails_on('sqlite', 'no schemas')
def test_get_primary_keys_with_schema(self):
- self._test_get_primary_keys(schema=get_schema())
+ self._test_get_primary_keys(schema='test_schema')
def _test_get_foreign_keys(self, schema=None):
meta = MetaData(testing.db)
@testing.requires.schemas
def test_get_foreign_keys_with_schema(self):
- self._test_get_foreign_keys(schema=get_schema())
+ self._test_get_foreign_keys(schema='test_schema')
def _test_get_indexes(self, schema=None):
meta = MetaData(testing.db)
@testing.requires.schemas
def test_get_indexes_with_schema(self):
- self._test_get_indexes(schema=get_schema())
+ self._test_get_indexes(schema='test_schema')
def _test_get_view_definition(self, schema=None):
meta = MetaData(testing.db)
@testing.requires.schemas
def test_get_view_definition_with_schema(self):
- self._test_get_view_definition(schema=get_schema())
+ self._test_get_view_definition(schema='test_schema')
def _test_get_table_oid(self, table_name, schema=None):
if testing.against('postgresql'):
@testing.requires.schemas
def test_get_table_oid_with_schema(self):
- self._test_get_table_oid('users', schema=get_schema())
+ self._test_get_table_oid('users', schema='test_schema')
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
