[ticket:2430]
- mysql
+ - [feature] Added support for MySQL index and
+ primary key constraint types
+ (i.e. USING) via new mysql_using parameter
+ to Index and PrimaryKeyConstraint,
+ courtesy Diana Clarke. [ticket:2386]
+
- [feature] Added support for the "isolation_level"
parameter to all MySQL dialects. Thanks
to mu_mind for the patch here. [ticket:2394]
an integer. MySQL only allows a length for an index if it is for a CHAR,
VARCHAR, TEXT, BINARY, VARBINARY and BLOB.
+Index Types
+~~~~~~~~~~~~~
+
+Some MySQL storage engines permit you to specify an index type when creating
+an index or primary key constraint. SQLAlchemy provides this feature via the
+``mysql_using`` parameter on :class:`.Index`::
+
+ Index('my_index', my_table.c.data, mysql_using='hash')
+
+As well as the ``mysql_using`` parameter on :class:`.PrimaryKeyConstraint`::
+
+ PrimaryKeyConstraint("data", mysql_using='hash')
+
+The value passed to the keyword argument will be simply passed through to the
+underlying CREATE INDEX or PRIMARY KEY clause, so it *must* be a valid index
+type for your MySQL storage engine.
+
More information can be found at:
+
http://dev.mysql.com/doc/refman/5.0/en/create-index.html
+
+http://dev.mysql.com/doc/refman/5.0/en/create-table.html
+
"""
import datetime, inspect, re, sys
table_opts.append(joiner.join((opt, arg)))
return ' '.join(table_opts)
+
def visit_create_index(self, create):
index = create.element
preparer = self.preparer
+ table = preparer.format_table(index.table)
+ columns = [preparer.quote(c.name, c.quote) for c in index.columns]
+ name = preparer.quote(
+ self._index_identifier(index.name),
+ index.quote)
+
text = "CREATE "
if index.unique:
text += "UNIQUE "
- text += "INDEX %s ON %s " \
- % (preparer.quote(self._index_identifier(index.name),
- index.quote),preparer.format_table(index.table))
+ text += "INDEX %s ON %s " % (name, table)
+
+ columns = ', '.join(columns)
if 'mysql_length' in index.kwargs:
length = index.kwargs['mysql_length']
+ text += "(%s(%d))" % (columns, length)
else:
- length = None
- if length is not None:
- text+= "(%s(%d))" \
- % (', '.join(preparer.quote(c.name, c.quote)
- for c in index.columns), length)
- else:
- text+= "(%s)" \
- % (', '.join(preparer.quote(c.name, c.quote)
- for c in index.columns))
+ text += "(%s)" % (columns)
+
+ if 'mysql_using' in index.kwargs:
+ using = index.kwargs['mysql_using']
+ text += " USING %s" % (preparer.quote(using, index.quote))
+
return text
+ def visit_primary_key_constraint(self, constraint):
+ text = super(MySQLDDLCompiler, self).\
+ visit_primary_key_constraint(constraint)
+ if "mysql_using" in constraint.kwargs:
+ using = constraint.kwargs['mysql_using']
+ text += " USING %s" % (
+ self.preparer.quote(using, constraint.quote))
+ return text
def visit_drop_index(self, drop):
index = drop.element
return "\nDROP INDEX %s ON %s" % \
- (self.preparer.quote(self._index_identifier(index.name), index.quote),
+ (self.preparer.quote(
+ self._index_identifier(index.name), index.quote
+ ),
self.preparer.format_table(index.table))
def visit_drop_constraint(self, drop):
else:
return schema + "." + name
+def _validate_dialect_kwargs(kwargs, name):
+ # validate remaining kwargs that they all specify DB prefixes
+ if len([k for k in kwargs
+ if not re.match(
+ r'^(?:%s)_' %
+ '|'.join(dialects.__all__), k
+ )
+ ]):
+ raise TypeError(
+ "Invalid argument(s) for %s: %r" % (name, kwargs.keys()))
+
class Table(SchemaItem, expression.TableClause):
"""Represent a table in a database.
def _extra_kwargs(self, **kwargs):
# validate remaining kwargs that they all specify DB prefixes
- if len([k for k in kwargs
- if not re.match(
- r'^(?:%s)_' %
- '|'.join(dialects.__all__), k
- )
- ]):
- raise TypeError(
- "Invalid argument(s) for Table: %r" % kwargs.keys())
+ _validate_dialect_kwargs(kwargs, "Table")
self.kwargs.update(kwargs)
def _init_collections(self):
__visit_name__ = 'constraint'
def __init__(self, name=None, deferrable=None, initially=None,
- _create_rule=None):
+ _create_rule=None,
+ **kw):
"""Create a SQL constraint.
:param name:
_create_rule is used by some types to create constraints.
Currently, its call signature is subject to change at any time.
+
+ :param \**kwargs:
+ Dialect-specific keyword parameters, see the documentation
+ for various dialects and constraints regarding options here.
"""
self.initially = initially
self._create_rule = _create_rule
util.set_creation_order(self)
+ _validate_dialect_kwargs(kw, self.__class__.__name__)
+ self.kwargs = kw
@property
def table(self):
self.assert_compile(x,
'''SELECT mysql_table.col1, mysql_table.`master_ssl_verify_server_cert` FROM mysql_table''')
- def test_create_index_with_length(self):
+ def test_create_index_simple(self):
m = MetaData()
tbl = Table('testtbl', m, Column('data', String(255)))
- idx = Index('test_idx1', tbl.c.data,
- mysql_length=10)
- idx2 = Index('test_idx2', tbl.c.data,
- mysql_length=5)
+ idx = Index('test_idx1', tbl.c.data)
self.assert_compile(schema.CreateIndex(idx),
- 'CREATE INDEX test_idx1 ON testtbl (data(10))',
- dialect=mysql.dialect())
+ 'CREATE INDEX test_idx1 ON testtbl (data)',
+ dialect=mysql.dialect())
+
+ def test_create_index_with_length(self):
+ m = MetaData()
+ tbl = Table('testtbl', m, Column('data', String(255)))
+ idx1 = Index('test_idx1', tbl.c.data, mysql_length=10)
+ idx2 = Index('test_idx2', tbl.c.data, mysql_length=5)
+
+ self.assert_compile(schema.CreateIndex(idx1),
+ 'CREATE INDEX test_idx1 ON testtbl (data(10))',
+ dialect=mysql.dialect())
+ self.assert_compile(schema.CreateIndex(idx2),
+ 'CREATE INDEX test_idx2 ON testtbl (data(5))',
+ dialect=mysql.dialect())
+
+ def test_create_index_with_using(self):
+ m = MetaData()
+ tbl = Table('testtbl', m, Column('data', String(255)))
+ idx1 = Index('test_idx1', tbl.c.data, mysql_using='btree')
+ idx2 = Index('test_idx2', tbl.c.data, mysql_using='hash')
+
+ self.assert_compile(schema.CreateIndex(idx1),
+ 'CREATE INDEX test_idx1 ON testtbl (data) USING btree',
+ dialect=mysql.dialect())
self.assert_compile(schema.CreateIndex(idx2),
- "CREATE INDEX test_idx2 ON testtbl (data(5))",
- dialect=mysql.dialect())
+ 'CREATE INDEX test_idx2 ON testtbl (data) USING hash',
+ dialect=mysql.dialect())
+
+ def test_create_pk_plain(self):
+ m = MetaData()
+ tbl = Table('testtbl', m, Column('data', String(255)),
+ PrimaryKeyConstraint('data'))
+
+ self.assert_compile(schema.CreateTable(tbl),
+ "CREATE TABLE testtbl (data VARCHAR(255), PRIMARY KEY (data))",
+ dialect=mysql.dialect())
+
+ def test_create_pk_with_using(self):
+ m = MetaData()
+ tbl = Table('testtbl', m, Column('data', String(255)),
+ PrimaryKeyConstraint('data', mysql_using='btree'))
+
+ self.assert_compile(schema.CreateTable(tbl),
+ "CREATE TABLE testtbl (data VARCHAR(255), "
+ "PRIMARY KEY (data) USING btree)",
+ dialect=mysql.dialect())
class DialectTest(fixtures.TestBase):
__only_on__ = 'mysql'
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
