2
from . import exclusions
3
from .. import schema, event
6
__all__ = 'Table', 'Column',
11
def Table(*args, **kw):
12
"""A schema.Table wrapper/hook for dialect-specific tweaks."""
14
test_opts = dict([(k, kw.pop(k)) for k in kw.keys()
15
if k.startswith('test_')])
17
kw.update(table_options)
19
if exclusions.against('mysql'):
20
if 'mysql_engine' not in kw and 'mysql_type' not in kw:
21
if 'test_needs_fk' in test_opts or 'test_needs_acid' in test_opts:
22
kw['mysql_engine'] = 'InnoDB'
24
kw['mysql_engine'] = 'MyISAM'
26
# Apply some default cascading rules for self-referential foreign keys.
27
# MySQL InnoDB has some issues around seleting self-refs too.
28
if exclusions.against('firebird'):
30
unpack = (config.db.dialect.
31
identifier_preparer.unformat_identifiers)
33
# Only going after ForeignKeys in Columns. May need to
34
# expand to ForeignKeyConstraint too.
36
for col in args if isinstance(col, schema.Column)
37
for fk in col.foreign_keys]
40
# root around in raw spec
42
if isinstance(ref, schema.Column):
45
# take just the table name: on FB there cannot be
46
# a schema, so the first element is always the
47
# table name, possibly followed by the field name
49
if name == table_name:
50
if fk.ondelete is None:
51
fk.ondelete = 'CASCADE'
52
if fk.onupdate is None:
53
fk.onupdate = 'CASCADE'
55
return schema.Table(*args, **kw)
58
def Column(*args, **kw):
59
"""A schema.Column wrapper/hook for dialect-specific tweaks."""
61
test_opts = dict([(k, kw.pop(k)) for k in kw.keys()
62
if k.startswith('test_')])
64
if not config.requirements.foreign_key_ddl.enabled:
65
args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)]
67
col = schema.Column(*args, **kw)
68
if 'test_needs_autoincrement' in test_opts and \
69
kw.get('primary_key', False):
71
# allow any test suite to pick up on this
72
col.info['test_needs_autoincrement'] = True
74
# hardcoded rule for firebird, oracle; this should
76
if exclusions.against('firebird', 'oracle'):
79
schema.Sequence(_truncate_name(
80
config.db.dialect, tbl.name + '_' + c.name + '_seq'),
83
event.listen(col, 'after_parent_attach', add_seq, propagate=True)
90
def _truncate_name(dialect, name):
91
if len(name) > dialect.max_identifier_length:
92
return name[0:max(dialect.max_identifier_length - 6, 0)] + \
93
"_" + hex(hash(name) % 64)[2:]