Commits

Mike Bayer  committed 4b3475b

- create_all(), drop_all(), create(), drop() all raise
an error if the table name or schema name contains
more characters than that dialect's configured
character limit. Some DB's can handle too-long
table names during usage, and SQLA can handle this
as well. But various reflection/
checkfirst-during-create scenarios fail since we are
looking for the name within the DB's catalog tables.
[ticket:571]

  • Participants
  • Parent commits 0298783
  • Branches rel_0_4

Comments (0)

Files changed (5)

       a transaction is in progress [ticket:976].  This
       flag is always True with a "transactional" 
       (in 0.5 a non-"autocommit") Session.
-    
+
+- schema
+    - create_all(), drop_all(), create(), drop() all raise
+      an error if the table name or schema name contains
+      more characters than that dialect's configured
+      character limit.  Some DB's can handle too-long
+      table names during usage, and SQLA can handle this
+      as well. But various reflection/
+      checkfirst-during-create scenarios fail since we are
+      looking for the name within the DB's catalog tables.
+      [ticket:571]
+
 - postgres
     - Repaired server_side_cursors to properly detect 
       text() clauses.
-      
+
 - mysql
     - Added 'CALL' to the list of SQL keywords which return
       result rows.

File lib/sqlalchemy/engine/default.py

 import re, random
 from sqlalchemy.engine import base
 from sqlalchemy.sql import compiler, expression
-
+from sqlalchemy import exceptions
 
 AUTOCOMMIT_REGEXP = re.compile(r'\s*(?:UPDATE|INSERT|CREATE|DELETE|DROP|ALTER)',
                                re.I | re.UNICODE)
             typeobj = typeobj()
         return typeobj
 
-
+    def validate_identifier(self, ident):
+        if len(ident) > self.max_identifier_length:
+            raise exceptions.IdentifierError("Identifier '%s' exceeds maximum length of %d characters" % (ident, self.max_identifier_length))
+        
     def oid_column_name(self, column):
         return None
 

File lib/sqlalchemy/exceptions.py

 class CircularDependencyError(SQLAlchemyError):
     """Raised by topological sorts when a circular dependency is detected"""
 
-
+class IdentifierError(SQLAlchemyError):
+    """Raised when a schema name is beyond the max character limit"""
+    
 class FlushError(SQLAlchemyError):
     """Raised when an invalid condition is detected upon a ``flush()``."""
 

File lib/sqlalchemy/sql/compiler.py

     def get_column_specification(self, column, first_pk=False):
         raise NotImplementedError()
 
+    def _can_create(self, table):
+        self.dialect.validate_identifier(table.name)
+        if table.schema:
+            self.dialect.validate_identifier(table.schema)
+        return not self.checkfirst or not self.dialect.has_table(self.connection, table.name, schema=table.schema)
+
     def visit_metadata(self, metadata):
-        collection = [t for t in metadata.table_iterator(reverse=False, tables=self.tables) if (not self.checkfirst or not self.dialect.has_table(self.connection, t.name, schema=t.schema))]
+        collection = [t for t in metadata.table_iterator(reverse=False, tables=self.tables) if self._can_create(t)]
         for table in collection:
             self.traverse_single(table)
         if self.dialect.supports_alter:
         self.dialect = dialect
 
     def visit_metadata(self, metadata):
-        collection = [t for t in metadata.table_iterator(reverse=True, tables=self.tables) if (not self.checkfirst or  self.dialect.has_table(self.connection, t.name, schema=t.schema))]
+        collection = [t for t in metadata.table_iterator(reverse=True, tables=self.tables) if self._can_drop(t)]
         if self.dialect.supports_alter:
             for alterable in self.find_alterables(collection):
                 self.drop_foreignkey(alterable)
         for table in collection:
             self.traverse_single(table)
 
+    def _can_drop(self, table):
+        self.dialect.validate_identifier(table.name)
+        if table.schema:
+            self.dialect.validate_identifier(table.schema)
+        return not self.checkfirst or self.dialect.has_table(self.connection, table.name, schema=table.schema)
+
     def visit_index(self, index):
         self.append("\nDROP INDEX " + self.preparer.format_index(index))
         self.execute()

File test/sql/labels.py

 import testenv; testenv.configure_for_tests()
 from sqlalchemy import *
+from sqlalchemy import exceptions
 from testlib import *
 from sqlalchemy.engine import default
 
         metadata.drop_all()
         testing.db.dialect.max_identifier_length = maxlen
 
+    def test_too_long_name_disallowed(self):
+        m = MetaData(testing.db)
+        t1 = Table("this_name_is_too_long_for_what_were_doing_in_this_test", m, Column('foo', Integer))
+        self.assertRaises(exceptions.IdentifierError, m.create_all)
+        self.assertRaises(exceptions.IdentifierError, m.drop_all)
+        self.assertRaises(exceptions.IdentifierError, t1.create)
+        self.assertRaises(exceptions.IdentifierError, t1.drop)
+        
     def test_result(self):
         table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"})
         table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"})