apply patch for Issue #43 (better SQLite support) by Florian Apolloner

This commit is contained in:
jan.dittberner 2008-12-30 20:43:40 +00:00
parent b73aa25733
commit 05819a6f70
2 changed files with 44 additions and 34 deletions

View File

@ -1,29 +1,52 @@
from migrate.changeset import ansisql,constraint,exceptions
from sqlalchemy.databases import sqlite as sa_base
from sqlalchemy import Table, MetaData
#import sqlalchemy as sa
SQLiteSchemaGenerator = sa_base.SQLiteSchemaGenerator
class SQLiteHelper(object):
def visit_column(self, param):
try:
table = self._to_table(param.table)
except:
table = self._to_table(param)
raise
table_name = self._to_table_name(table)
self.append('ALTER TABLE %s RENAME TO migration_tmp' % table_name)
self.execute()
insertion_string = self._modify_table(table, param)
table.create()
self.append(insertion_string % {'table_name': table_name})
self.execute()
self.append('DROP TABLE migration_tmp')
self.execute()
class SQLiteColumnGenerator(SQLiteSchemaGenerator,ansisql.ANSIColumnGenerator):
pass
class SQLiteColumnDropper(ansisql.ANSIColumnDropper):
def visit_column(self,column):
raise exceptions.NotSupportedError("SQLite does not support "
"DROP COLUMN; see http://www.sqlite.org/lang_altertable.html")
class SQLiteSchemaChanger(ansisql.ANSISchemaChanger):
class SQLiteColumnDropper(SQLiteHelper, ansisql.ANSIColumnDropper):
def _modify_table(self,table, column):
del table.columns[column.name]
columns = ','.join([c.name for c in table.columns])
return 'INSERT INTO %(table_name)s SELECT ' + columns + ' from migration_tmp'
class SQLiteSchemaChanger(SQLiteHelper, ansisql.ANSISchemaChanger):
def _not_supported(self,op):
raise exceptions.NotSupportedError("SQLite does not support "
"%s; see http://www.sqlite.org/lang_altertable.html"%op)
def _visit_column_nullable(self,table_name,col_name,delta):
return self._not_supported('ALTER TABLE')
def _visit_column_default(self,table_name,col_name,delta):
return self._not_supported('ALTER TABLE')
def _visit_column_type(self,table_name,col_name,delta):
return self._not_supported('ALTER TABLE')
def _visit_column_name(self,table_name,col_name,delta):
return self._not_supported('ALTER TABLE')
def _modify_table(self, table, delta):
column = table.columns[delta.current_name]
for k,v in delta.items():
setattr(column, k, v)
return 'INSERT INTO %(table_name)s SELECT * from migration_tmp'
def visit_index(self,param):
self._not_supported('ALTER INDEX')
def _do_quote_column_identifier(self, identifier):
return '"%s"'%identifier
@ -36,6 +59,7 @@ class SQLiteConstraintGenerator(ansisql.ANSIConstraintGenerator):
msg = tmpl%(name,tname,cols)
self.append(msg)
self.execute()
class SQLiteConstraintDropper(ansisql.ANSIColumnDropper):
def visit_migrate_primary_key_constraint(self,constraint):
tmpl = "DROP INDEX %s "

View File

@ -66,10 +66,6 @@ class TestAddDropColumn(fixture.DB):
self.assertEquals(getattr(self.table.c,col_name),col)
#drop_column(col,self.table)
col = getattr(self.table.c,col_name)
# SQLite can't do drop column: stop here
if self.url.startswith('sqlite://'):
self.assertRaises(changeset.exceptions.NotSupportedError,drop_column_func,col)
return
drop_column_func(col)
assert_numcols(1)
@ -301,18 +297,7 @@ class TestColumnChange(fixture.DB):
#self.engine.echo=False
super(TestColumnChange, self)._teardown()
@fixture.usedb(supported='sqlite')
def test_sqlite_not_supported(self):
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,server_default=DefaultClause('tluafed'))
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,nullable=True)
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,type=String(21))
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,name='atad')
@fixture.usedb(not_supported='sqlite')
@fixture.usedb()
def test_rename(self):
"""Can rename a column"""
def num_rows(col,content):
@ -354,7 +339,7 @@ class TestColumnChange(fixture.DB):
self.table.c.data # Should not raise exception
self.assertEquals(num_rows(self.table.c.data,content),1)
@fixture.usedb(not_supported='sqlite')
@fixture.usedb()
def xtest_fk(self):
"""Can add/drop foreign key constraints to/from a column
Not supported
@ -371,7 +356,7 @@ class TestColumnChange(fixture.DB):
self.refresh_table(self.table.name)
self.assert_(self.table.c.data.foreign_key is None)
@fixture.usedb(not_supported='sqlite')
@fixture.usedb()
def test_type(self):
"""Can change a column's type"""
# Entire column definition given
@ -394,7 +379,7 @@ class TestColumnChange(fixture.DB):
self.refresh_table(self.table.name)
self.assert_(isinstance(self.table.c.id.type,String))
@fixture.usedb(not_supported=('sqlite', 'mysql'))
@fixture.usedb(not_supported='mysql')
def test_default(self):
"""Can change a column's server_default value (DefaultClauses only)
Only DefaultClauses are changed here: others are managed by the
@ -427,7 +412,7 @@ class TestColumnChange(fixture.DB):
self.assert_(row['data'] is None,row['data'])
@fixture.usedb(not_supported='sqlite')
@fixture.usedb()
def test_null(self):
"""Can change a column's null constraint"""
self.assertEquals(self.table.c.data.nullable,True)
@ -443,11 +428,12 @@ class TestColumnChange(fixture.DB):
self.refresh_table(self.table.name)
self.assertEquals(self.table.c.data.nullable,True)
@fixture.usedb(not_supported='sqlite')
@fixture.usedb()
def xtest_pk(self):
"""Can add/drop a column to/from its table's primary key
Not supported
"""
self.engine.echo = True
self.assertEquals(len(self.table.primary_key),1)
# Entire column definition