1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
"""
MySQL database specific implementations of changeset classes.
"""
from migrate.changeset import ansisql, exceptions, SQLA_06
from sqlalchemy.databases import mysql as sa_base
from sqlalchemy import types as sqltypes
if not SQLA_06:
MySQLSchemaGenerator = sa_base.MySQLSchemaGenerator
else:
MySQLSchemaGenerator = sa_base.MySQLDDLCompiler
class MySQLColumnGenerator(MySQLSchemaGenerator, ansisql.ANSIColumnGenerator):
pass
class MySQLColumnDropper(ansisql.ANSIColumnDropper):
pass
class MySQLSchemaChanger(MySQLSchemaGenerator, ansisql.ANSISchemaChanger):
def visit_column(self, delta):
table = delta.table
colspec = self.get_column_specification(delta.result_column)
if delta.result_column.autoincrement:
primary_keys = [c for c in table.primary_key.columns
if (c.autoincrement and
isinstance(c.type, sqltypes.Integer) and
not c.foreign_keys)]
if primary_keys:
first = primary_keys.pop(0)
if first.name == delta.current_name:
colspec += " AUTO_INCREMENT"
old_col_name = self.preparer.quote(delta.current_name, table.quote)
self.start_alter_table(table)
self.append("CHANGE COLUMN %s " % old_col_name)
self.append(colspec)
self.execute()
def visit_index(self, param):
# If MySQL can do this, I can't find how
raise exceptions.NotSupportedError("MySQL cannot rename indexes")
class MySQLConstraintGenerator(ansisql.ANSIConstraintGenerator):
pass
if SQLA_06:
class MySQLConstraintDropper(MySQLSchemaGenerator, ansisql.ANSIConstraintDropper):
def visit_migrate_check_constraint(self, *p, **k):
raise exceptions.NotSupportedError("MySQL does not support CHECK"
" constraints, use triggers instead.")
else:
class MySQLConstraintDropper(ansisql.ANSIConstraintDropper):
def visit_migrate_primary_key_constraint(self, constraint):
self.start_alter_table(constraint)
self.append("DROP PRIMARY KEY")
self.execute()
def visit_migrate_foreign_key_constraint(self, constraint):
self.start_alter_table(constraint)
self.append("DROP FOREIGN KEY ")
constraint.name = self.get_constraint_name(constraint)
self.append(self.preparer.format_constraint(constraint))
self.execute()
def visit_migrate_check_constraint(self, *p, **k):
raise exceptions.NotSupportedError("MySQL does not support CHECK"
" constraints, use triggers instead.")
def visit_migrate_unique_constraint(self, constraint, *p, **k):
self.start_alter_table(constraint)
self.append('DROP INDEX ')
constraint.name = self.get_constraint_name(constraint)
self.append(self.preparer.format_constraint(constraint))
self.execute()
class MySQLDialect(ansisql.ANSIDialect):
columngenerator = MySQLColumnGenerator
columndropper = MySQLColumnDropper
schemachanger = MySQLSchemaChanger
constraintgenerator = MySQLConstraintGenerator
constraintdropper = MySQLConstraintDropper
|
