From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Sat, 24 May 2008 23:34:04 +0000 (+0000)
Subject: oracle dialect takes schema name into account when checking for existing tables
X-Git-Tag: rel_0_5beta1~23
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=fa42abd21323f9be5bbf9f8744e02271c9834691;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

oracle dialect takes schema name into account when checking for existing tables
of the same name. [ticket:709]
---

diff --git a/CHANGES b/CHANGES
index f5ee3da0a9..31f6bd6687 100644
--- a/CHANGES
+++ b/CHANGES
@@ -77,6 +77,12 @@ user_defined_state
       before returning, meaning it returns a lower-case name
       when the identifier is detected as case insensitive.
 
+    - creating/dropping tables takes schema name into account
+      when searching for the existing table, so that tables
+      in other owner namespaces with the same name do not
+      conflict [ticket:709]
+
+
 0.4.6
 =====
 - orm
diff --git a/lib/sqlalchemy/databases/oracle.py b/lib/sqlalchemy/databases/oracle.py
index ed1778a6f9..355012a1d6 100644
--- a/lib/sqlalchemy/databases/oracle.py
+++ b/lib/sqlalchemy/databases/oracle.py
@@ -358,7 +358,9 @@ class OracleDialect(default.DefaultDialect):
         return OracleExecutionContext(self, *args, **kwargs)
 
     def has_table(self, connection, table_name, schema=None):
-        cursor = connection.execute("""select table_name from all_tables where table_name=:name""", {'name':self._denormalize_name(table_name)})
+        if not schema:
+            schema = self.get_default_schema_name(connection)
+        cursor = connection.execute("""select table_name from all_tables where table_name=:name and owner=:schema_name""", {'name':self._denormalize_name(table_name), 'schema_name':self._denormalize_name(schema)})
         return cursor.fetchone() is not None
 
     def has_sequence(self, connection, sequence_name):
diff --git a/test/dialect/oracle.py b/test/dialect/oracle.py
index 24353152d3..5ca906c357 100644
--- a/test/dialect/oracle.py
+++ b/test/dialect/oracle.py
@@ -156,7 +156,7 @@ WHERE ora_rn>5 AND ora_rn<=15", dialect=oracle.dialect(use_ansi=False))
             "ON addresses.address_type_id = address_types_1.id WHERE addresses.user_id = :user_id_1 ORDER BY addresses.rowid, "
             "address_types.rowid")
 
-class SchemaReflectionTest(TestBase, AssertsCompiledSQL):
+class MultiSchemaTest(TestBase, AssertsCompiledSQL):
     """instructions:
 
        1. create a user 'ed' in the oracle database.
@@ -174,6 +174,44 @@ class SchemaReflectionTest(TestBase, AssertsCompiledSQL):
 
     __only_on__ = 'oracle'
 
+    def test_create_same_names_explicit_schema(self):
+        schema = testing.db.dialect.get_default_schema_name(testing.db.connect())
+        meta = MetaData(testing.db)
+        parent = Table('parent', meta, 
+            Column('pid', Integer, primary_key=True),
+            schema=schema
+        )
+        child = Table('child', meta, 
+            Column('cid', Integer, primary_key=True),
+            Column('pid', Integer, ForeignKey('scott.parent.pid')),
+            schema=schema
+        )
+        meta.create_all()
+        try:
+            parent.insert().execute({'pid':1})
+            child.insert().execute({'cid':1, 'pid':1})
+            self.assertEquals(child.select().execute().fetchall(), [(1, 1)])
+        finally:
+            meta.drop_all()
+
+    def test_create_same_names_implicit_schema(self):
+        meta = MetaData(testing.db)
+        parent = Table('parent', meta, 
+            Column('pid', Integer, primary_key=True),
+        )
+        child = Table('child', meta, 
+            Column('cid', Integer, primary_key=True),
+            Column('pid', Integer, ForeignKey('parent.pid')),
+        )
+        meta.create_all()
+        try:
+            parent.insert().execute({'pid':1})
+            child.insert().execute({'cid':1, 'pid':1})
+            self.assertEquals(child.select().execute().fetchall(), [(1, 1)])
+        finally:
+            meta.drop_all()
+
+
     def test_reflect_alt_owner_explicit(self):
         meta = MetaData(testing.db)
         parent = Table('parent', meta, autoload=True, schema='ed')