x/sqlalchemy-migrate
Code Issues Proposed changes

add support for SA 0.5

Browse Source
This commit is contained in:
christian.simms 2008-08-28 01:17:44 +00:00
parent aba0f50681
commit f687cd3fe0
6 changed files with 24 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
migrate/changeset/ansisql.py
View File

@ -124,7 +124,8 @@ class ANSISchemaChanger(AlterTableVisitor,SchemaGenerator):
self._run_subvisit(delta,self._visit_column_type)
if 'nullable' in keys:
self._run_subvisit(delta,self._visit_column_nullable)
if 'default' in keys:
if 'server_default' in keys:
# Skip 'default': only handle server-side defaults, others are managed by the app, not the db.
self._run_subvisit(delta,self._visit_column_default)
#if 'primary_key' in keys:
# #self._run_subvisit(delta,self._visit_column_primary_key)
@ -172,14 +173,9 @@ class ANSISchemaChanger(AlterTableVisitor,SchemaGenerator):
else:
self.append("SET NOT NULL")
def _visit_column_default(self,table_name,col_name,delta):
default = delta['default']
# Default must be a PassiveDefault; else, ignore
# (Non-PassiveDefaults are managed by the app, not the db)
if default is not None:
if not isinstance(default,sa.PassiveDefault):
return
server_default = delta['server_default']
# Dummy column: get_col_default_string needs a column for some reason
dummy = sa.Column(None,None,default=default)
dummy = sa.Column(None,None,server_default=server_default)
default_text = self.get_column_default_string(dummy)
self.start_alter_table(table_name)
self.append("ALTER COLUMN %s "%col_name)

2
migrate/changeset/databases/mysql.py
View File

@ -13,7 +13,7 @@ class MySQLSchemaChanger(MySQLSchemaGenerator,ansisql.ANSISchemaChanger):
keys = delta.keys()
if 'type' in keys or 'nullable' in keys or 'name' in keys:
self._run_subvisit(delta,self._visit_column_change)
if 'default' in keys:
if 'server_default' in keys:
# Column name might have changed above
col_name = delta.get('name',delta.current_name)
self._run_subvisit(delta,self._visit_column_default,col_name=col_name)

8
migrate/changeset/databases/oracle.py
View File

@ -22,7 +22,7 @@ class OracleSchemaChanger(OracleSchemaGenerator,ansisql.ANSISchemaChanger):
def visit_column(self,delta):
keys = delta.keys()
if 'type' in keys or 'nullable' in keys or 'default' in keys:
if 'type' in keys or 'nullable' in keys or 'default' in keys or 'server_default' in keys:
self._run_subvisit(delta,self._visit_column_change)
if 'name' in keys:
self._run_subvisit(delta,self._visit_column_name)
@ -36,7 +36,7 @@ class OracleSchemaChanger(OracleSchemaGenerator,ansisql.ANSISchemaChanger):
# Oracle cannot drop a default once created, but it can set it to null.
# We'll do that if default=None
# http://forums.oracle.com/forums/message.jspa?messageID=1273234#1273234
dropdefault_hack = (column.default is None and 'default' in delta.keys())
dropdefault_hack = (column.server_default is None and 'server_default' in delta.keys())
# Oracle apparently doesn't like it when we say "not null" if the
# column's already not null. Fudge it, so we don't need a new function
notnull_hack = ((not column.nullable) and ('nullable' not in delta.keys()))
@ -45,7 +45,7 @@ class OracleSchemaChanger(OracleSchemaGenerator,ansisql.ANSISchemaChanger):
if dropdefault_hack:
column.default = sa.PassiveDefault(sa.sql.null())
column.server_default = sa.PassiveDefault(sa.sql.null())
if notnull_hack:
column.nullable = True
colspec=self.get_column_specification(column,override_nullable=null_hack)
@ -54,7 +54,7 @@ class OracleSchemaChanger(OracleSchemaGenerator,ansisql.ANSISchemaChanger):
if notnull_hack:
column.nullable = False
if dropdefault_hack:
column.default = None
column.server_default = None
self.start_alter_table(table_name)
self.append("MODIFY ")

4
migrate/changeset/schema.py
View File

@ -158,7 +158,7 @@ class _ColumnDelta(dict):
diffs = func(*p,**k)
self._set_diffs(diffs)
# Column attributes that can be altered
diff_keys = ('name','type','nullable','default','primary_key','foreign_key')
diff_keys = ('name','type','nullable','default','server_default','primary_key','foreign_key')
def _get_table_name(self):
if isinstance(self._table,basestring):
@ -198,7 +198,7 @@ class _ColumnDelta(dict):
self.result_column = start_col.copy()
self._table = k.pop('table',None) or start_col.table or end_col.table
self.current_name = start_col.name
for key in ('name','nullable','default','primary_key','foreign_key'):
for key in ('name','nullable','default','server_default','primary_key','foreign_key'):
val = getattr(end_col,key,None)
if getattr(start_col,key,None) != val:
k.setdefault(key,val)

24
test/changeset/test_changeset.py
View File

@ -57,7 +57,7 @@ class TestAddDropColumn(fixture.DB):
_assert_numcols(expected,'database')
assert_numcols(1)
if len(col_p) == 0:
col_p = [String]
col_p = [String(40)]
col = Column(col_name,*col_p,**col_k)
create_column_func(col)
#create_column(col,self.table)
@ -306,7 +306,7 @@ class TestColumnChange(fixture.DB):
@fixture.usedb(supported='sqlite')
def test_sqlite_not_supported(self):
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,default=PassiveDefault('tluafed'))
self.table.c.data.alter,server_default=PassiveDefault('tluafed'))
self.assertRaises(changeset.exceptions.NotSupportedError,
self.table.c.data.alter,nullable=True)
self.assertRaises(changeset.exceptions.NotSupportedError,
@ -398,32 +398,32 @@ class TestColumnChange(fixture.DB):
@fixture.usedb(not_supported='sqlite')
def test_default(self):
"""Can change a column's default value (PassiveDefaults only)
"""Can change a column's server_default value (PassiveDefaults only)
Only PassiveDefaults are changed here: others are managed by the
application / by SA
"""
#self.engine.echo=True
self.assertEquals(self.table.c.data.default.arg,'tluafed')
self.assertEquals(self.table.c.data.server_default.arg,'tluafed')
# Just the new default
default = 'my_default'
self.table.c.data.alter(default=PassiveDefault(default))
self.table.c.data.alter(server_default=PassiveDefault(default))
self.refresh_table(self.table.name)
#self.assertEquals(self.table.c.data.default.arg,default)
#self.assertEquals(self.table.c.data.server_default.arg,default)
# TextClause returned by autoload
self.assert_(default in str(self.table.c.data.default.arg))
self.assert_(default in str(self.table.c.data.server_default.arg))
# Column object
default = 'your_default'
self.table.c.data.alter(Column('data',String(40),default=PassiveDefault(default)))
self.table.c.data.alter(Column('data',String(40),server_default=PassiveDefault(default)))
self.refresh_table(self.table.name)
self.assert_(default in str(self.table.c.data.default.arg))
self.assert_(default in str(self.table.c.data.server_default.arg))
# Remove default
self.table.c.data.alter(default=None)
self.table.c.data.alter(server_default=None)
self.refresh_table(self.table.name)
# default isn't necessarily None for Oracle
#self.assert_(self.table.c.data.default is None,self.table.c.data.default)
# server_default isn't necessarily None for Oracle
#self.assert_(self.table.c.data.server_default is None,self.table.c.data.server_default)
self.engine.execute(self.table.insert(),id=11)
row = self.table.select().execute().fetchone()
self.assert_(row['data'] is None,row['data'])

2
test/fixture/database.py
View File

@ -105,7 +105,7 @@ class DB(Base):
self.session = create_session(bind=self.engine)
if self.level < self.TXN:
return
self.txn = self.session.create_transaction()
self.txn = self.session.begin()
#self.txn.add(self.engine)
def _disconnect(self):