diff options
| -rw-r--r-- | lib/sqlalchemy/databases/mssql.py | 8 | ||||
| -rw-r--r-- | lib/sqlalchemy/databases/sqlite.py | 6 | ||||
| -rw-r--r-- | test/engine/reflection.py | 17 | ||||
| -rw-r--r-- | test/sql/query.py | 16 |
4 files changed, 38 insertions, 9 deletions
diff --git a/lib/sqlalchemy/databases/mssql.py b/lib/sqlalchemy/databases/mssql.py index 0e5a41a34..3f181c5bb 100644 --- a/lib/sqlalchemy/databases/mssql.py +++ b/lib/sqlalchemy/databases/mssql.py @@ -246,8 +246,7 @@ class MSSQLExecutionContext(default.DefaultExecutionContext): self.IINSERT = False if self.IINSERT: - # TODO: quoting rules for table name here ? - self.cursor.execute("SET IDENTITY_INSERT %s ON" % self.compiled.statement.table.fullname) + self.cursor.execute("SET IDENTITY_INSERT %s ON" % self.dialect.preparer().format_table(self.compiled.statement.table)) super(MSSQLExecutionContext, self).pre_exec() @@ -259,8 +258,7 @@ class MSSQLExecutionContext(default.DefaultExecutionContext): if self.compiled.isinsert: if self.IINSERT: - # TODO: quoting rules for table name here ? - self.cursor.execute("SET IDENTITY_INSERT %s OFF" % self.compiled.statement.table.fullname) + self.cursor.execute("SET IDENTITY_INSERT %s OFF" % self.dialect.preparer().format_table(self.compiled.statement.table)) self.IINSERT = False elif self.HASIDENT: if not len(self._last_inserted_ids) or self._last_inserted_ids[0] is None: @@ -531,7 +529,7 @@ class MSSQLDialect(ansisql.ANSIDialect): raise exceptions.NoSuchTableError(table.name) # We also run an sp_columns to check for identity columns: - cursor = connection.execute("sp_columns " + table.name) + cursor = connection.execute("sp_columns " + self.preparer().format_table(table)) ic = None while True: row = cursor.fetchone() diff --git a/lib/sqlalchemy/databases/sqlite.py b/lib/sqlalchemy/databases/sqlite.py index 7ccf38c4b..70cbd0c0e 100644 --- a/lib/sqlalchemy/databases/sqlite.py +++ b/lib/sqlalchemy/databases/sqlite.py @@ -235,7 +235,7 @@ class SQLiteDialect(ansisql.ANSIDialect): return (row is not None) def reflecttable(self, connection, table): - c = connection.execute("PRAGMA table_info(" + table.name + ")", {}) + c = connection.execute("PRAGMA table_info(%s)" % self.preparer().format_table(table), {}) found_table = False while True: row = c.fetchone() @@ -273,7 +273,7 @@ class SQLiteDialect(ansisql.ANSIDialect): if not found_table: raise exceptions.NoSuchTableError(table.name) - c = connection.execute("PRAGMA foreign_key_list(" + table.name + ")", {}) + c = connection.execute("PRAGMA foreign_key_list(%s)" % self.preparer().format_table(table), {}) fks = {} while True: row = c.fetchone() @@ -302,7 +302,7 @@ class SQLiteDialect(ansisql.ANSIDialect): for name, value in fks.iteritems(): table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1])) # check for UNIQUE indexes - c = connection.execute("PRAGMA index_list(" + table.name + ")", {}) + c = connection.execute("PRAGMA index_list(%s)" % self.preparer().format_table(table), {}) unique_indexes = [] while True: row = c.fetchone() diff --git a/test/engine/reflection.py b/test/engine/reflection.py index 75abf7ef8..85701599e 100644 --- a/test/engine/reflection.py +++ b/test/engine/reflection.py @@ -474,10 +474,25 @@ class ReflectionTest(PersistTest): meta2 = MetaData(testbase.db) try: table2 = Table('identity_test', meta2, autoload=True) - print table2.c['col1'].sequence + assert table2.c['col1'].sequence.start == 2 + assert table2.c['col1'].sequence.increment == 3 finally: table.drop() + def testreserved(self): + # check a table that uses an SQL reserved name doesn't cause an error + meta = MetaData(testbase.db) + table = Table( + 'select', meta, + Column('col1', Integer, primary_key=True) + ) + table.create() + + meta2 = MetaData(testbase.db) + try: + table2 = Table('select', meta2, autoload=True) + finally: + table.drop() class CreateDropTest(PersistTest): def setUpAll(self): diff --git a/test/sql/query.py b/test/sql/query.py index 6e4a76d11..772ffa793 100644 --- a/test/sql/query.py +++ b/test/sql/query.py @@ -443,6 +443,22 @@ class QueryTest(PersistTest): finally: tbl.drop() con.execute('drop schema paj') + + @testbase.supported('mssql') + def test_insertid_reserved(self): + meta = MetaData(testbase.db) + table = Table( + 'select', meta, + Column('col', Integer, primary_key=True) + ) + table.create() + + meta2 = MetaData(testbase.db) + try: + table.insert().execute(col=7) + finally: + table.drop() + def test_in_filtering(self): """test the behavior of the in_() function.""" |