__visit_name__ = 'LONGBLOB'
+class _EnumeratedValues(_StringType):
+ def _init_values(self, values, kw):
+ self.quoting = kw.pop('quoting', 'auto')
+
+ if self.quoting == 'auto' and len(values):
+ # What quoting character are we using?
+ q = None
+ for e in values:
+ if len(e) == 0:
+ self.quoting = 'unquoted'
+ break
+ elif q is None:
+ q = e[0]
+
+ if len(e) == 1 or e[0] != q or e[-1] != q:
+ self.quoting = 'unquoted'
+ break
+ else:
+ self.quoting = 'quoted'
+
+ if self.quoting == 'quoted':
+ util.warn_deprecated(
+ 'Manually quoting %s value literals is deprecated. Supply '
+ 'unquoted values and use the quoting= option in cases of '
+ 'ambiguity.' % self.__class__.__name__)
+
+ values = self._strip_values(values)
+
+ self._enumerated_values = values
+ length = max([len(v) for v in values] + [0])
+ return values, length
-class ENUM(sqltypes.Enum, _StringType):
+ @classmethod
+ def _strip_values(cls, values):
+ strip_values = []
+ for a in values:
+ if a[0:1] == '"' or a[0:1] == "'":
+ # strip enclosing quotes and unquote interior
+ a = a[1:-1].replace(a[0] * 2, a[0])
+ strip_values.append(a)
+ return strip_values
+
+class ENUM(sqltypes.Enum, _EnumeratedValues):
"""MySQL ENUM type."""
__visit_name__ = 'ENUM'
def __init__(self, *enums, **kw):
"""Construct an ENUM.
- Example:
+ E.g.::
- Column('myenum', MSEnum("foo", "bar", "baz"))
+ Column('myenum', ENUM("foo", "bar", "baz"))
:param enums: The range of valid values for this ENUM. Values will be
quoted when generating the schema according to the quoting flag (see
literals for you. This is a transitional option.
"""
- self.quoting = kw.pop('quoting', 'auto')
-
- if self.quoting == 'auto' and len(enums):
- # What quoting character are we using?
- q = None
- for e in enums:
- if len(e) == 0:
- self.quoting = 'unquoted'
- break
- elif q is None:
- q = e[0]
-
- if e[0] != q or e[-1] != q:
- self.quoting = 'unquoted'
- break
- else:
- self.quoting = 'quoted'
-
- if self.quoting == 'quoted':
- util.warn_deprecated(
- 'Manually quoting ENUM value literals is deprecated. Supply '
- 'unquoted values and use the quoting= option in cases of '
- 'ambiguity.')
- enums = self._strip_enums(enums)
-
+ values, length = self._init_values(enums, kw)
self.strict = kw.pop('strict', False)
- length = max([len(v) for v in enums] + [0])
kw.pop('metadata', None)
kw.pop('schema', None)
kw.pop('name', None)
kw.pop('native_enum', None)
kw.pop('inherit_schema', None)
_StringType.__init__(self, length=length, **kw)
- sqltypes.Enum.__init__(self, *enums)
-
- @classmethod
- def _strip_enums(cls, enums):
- strip_enums = []
- for a in enums:
- if a[0:1] == '"' or a[0:1] == "'":
- # strip enclosing quotes and unquote interior
- a = a[1:-1].replace(a[0] * 2, a[0])
- strip_enums.append(a)
- return strip_enums
+ sqltypes.Enum.__init__(self, *values)
def bind_processor(self, dialect):
super_convert = super(ENUM, self).bind_processor(dialect)
return sqltypes.Enum.adapt(self, impltype, **kw)
-class SET(_StringType):
+class SET(_EnumeratedValues):
"""MySQL SET type."""
__visit_name__ = 'SET'
def __init__(self, *values, **kw):
"""Construct a SET.
- Example::
+ E.g.::
- Column('myset', MSSet("'foo'", "'bar'", "'baz'"))
+ Column('myset', SET("foo", "bar", "baz"))
:param values: The range of valid values for this SET. Values will be
- used exactly as they appear when generating schemas. Strings must
- be quoted, as in the example above. Single-quotes are suggested for
- ANSI compatibility and are required for portability to servers with
- ANSI_QUOTES enabled.
+ quoted when generating the schema according to the quoting flag (see
+ below).
+
+ .. versionchanged:: 0.9.0 quoting is applied automatically to
+ :class:`.mysql.SET` in the same way as for :class:`.mysql.ENUM`.
:param charset: Optional, a column-level character set for this string
value. Takes precedence to 'ascii' or 'unicode' short-hand.
BINARY in schema. This does not affect the type of data stored,
only the collation of character data.
- """
- self._ddl_values = values
+ :param quoting: Defaults to 'auto': automatically determine enum value
+ quoting. If all enum values are surrounded by the same quoting
+ character, then use 'quoted' mode. Otherwise, use 'unquoted' mode.
- strip_values = []
- for a in values:
- if a[0:1] == '"' or a[0:1] == "'":
- # strip enclosing quotes and unquote interior
- a = a[1:-1].replace(a[0] * 2, a[0])
- strip_values.append(a)
+ 'quoted': values in enums are already quoted, they will be used
+ directly when generating the schema - this usage is deprecated.
+
+ 'unquoted': values in enums are not quoted, they will be escaped and
+ surrounded by single quotes when generating the schema.
- self.values = strip_values
- kw.setdefault('length', max([len(v) for v in strip_values] + [0]))
+ Previous versions of this type always required manually quoted
+ values to be supplied; future versions will always quote the string
+ literals for you. This is a transitional option.
+
+ .. versionadded:: 0.9.0
+
+ """
+ values, length = self._init_values(values, kw)
+ self.values = tuple(values)
+
+ kw.setdefault('length', length)
super(SET, self).__init__(**kw)
def result_processor(self, dialect, coltype):
if not type_.native_enum:
return super(MySQLTypeCompiler, self).visit_enum(type_)
else:
- return self.visit_ENUM(type_)
+ return self._visit_enumerated_values("ENUM", type_, type_.enums)
def visit_BLOB(self, type_):
if type_.length:
def visit_LONGBLOB(self, type_):
return "LONGBLOB"
- def visit_ENUM(self, type_):
+ def _visit_enumerated_values(self, name, type_, enumerated_values):
quoted_enums = []
- for e in type_.enums:
+ for e in enumerated_values:
quoted_enums.append("'%s'" % e.replace("'", "''"))
- return self._extend_string(type_, {}, "ENUM(%s)" %
- ",".join(quoted_enums))
+ return self._extend_string(type_, {}, "%s(%s)" % (
+ name, ",".join(quoted_enums))
+ )
+
+ def visit_ENUM(self, type_):
+ return self._visit_enumerated_values("ENUM", type_,
+ type_._enumerated_values)
def visit_SET(self, type_):
- return self._extend_string(type_, {}, "SET(%s)" %
- ",".join(type_._ddl_values))
+ return self._visit_enumerated_values("SET", type_,
+ type_._enumerated_values)
def visit_BOOLEAN(self, type):
return "BOOL"
if spec.get(kw, False):
type_kw[kw] = spec[kw]
- if type_ == 'enum':
- type_args = ENUM._strip_enums(type_args)
+ if issubclass(col_type, _EnumeratedValues):
+ type_args = _EnumeratedValues._strip_values(type_args)
type_instance = col_type(*type_args, **type_kw)
@testing.uses_deprecated('Manually quoting ENUM value literals')
def test_type_reflection(self):
# (ask_for, roundtripped_as_if_different)
- specs = [( String(1), mysql.MSString(1), ),
- ( String(3), mysql.MSString(3), ),
- ( Text(), mysql.MSText(), ),
- ( Unicode(1), mysql.MSString(1), ),
- ( Unicode(3), mysql.MSString(3), ),
- ( UnicodeText(), mysql.MSText(), ),
- ( mysql.MSChar(1), ),
- ( mysql.MSChar(3), ),
- ( NCHAR(2), mysql.MSChar(2), ),
- ( mysql.MSNChar(2), mysql.MSChar(2), ), # N is CREATE only
- ( mysql.MSNVarChar(22), mysql.MSString(22), ),
- ( SmallInteger(), mysql.MSSmallInteger(), ),
- ( SmallInteger(), mysql.MSSmallInteger(4), ),
- ( mysql.MSSmallInteger(), ),
- ( mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
- ( mysql.MSMediumInteger(), mysql.MSMediumInteger(), ),
- ( mysql.MSMediumInteger(8), mysql.MSMediumInteger(8), ),
- ( LargeBinary(3), mysql.TINYBLOB(), ),
- ( LargeBinary(), mysql.BLOB() ),
- ( mysql.MSBinary(3), mysql.MSBinary(3), ),
- ( mysql.MSVarBinary(3),),
- ( mysql.MSTinyBlob(),),
- ( mysql.MSBlob(),),
- ( mysql.MSBlob(1234), mysql.MSBlob()),
- ( mysql.MSMediumBlob(),),
- ( mysql.MSLongBlob(),),
- ( mysql.ENUM("''","'fleem'"), ),
+ specs = [(String(1), mysql.MSString(1), ),
+ (String(3), mysql.MSString(3), ),
+ (Text(), mysql.MSText(), ),
+ (Unicode(1), mysql.MSString(1), ),
+ (Unicode(3), mysql.MSString(3), ),
+ (UnicodeText(), mysql.MSText(), ),
+ (mysql.MSChar(1), ),
+ (mysql.MSChar(3), ),
+ (NCHAR(2), mysql.MSChar(2), ),
+ (mysql.MSNChar(2), mysql.MSChar(2), ), # N is CREATE only
+ (mysql.MSNVarChar(22), mysql.MSString(22), ),
+ (SmallInteger(), mysql.MSSmallInteger(), ),
+ (SmallInteger(), mysql.MSSmallInteger(4), ),
+ (mysql.MSSmallInteger(), ),
+ (mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
+ (mysql.MSMediumInteger(), mysql.MSMediumInteger(), ),
+ (mysql.MSMediumInteger(8), mysql.MSMediumInteger(8), ),
+ (LargeBinary(3), mysql.TINYBLOB(), ),
+ (LargeBinary(), mysql.BLOB() ),
+ (mysql.MSBinary(3), mysql.MSBinary(3), ),
+ (mysql.MSVarBinary(3),),
+ (mysql.MSTinyBlob(),),
+ (mysql.MSBlob(),),
+ (mysql.MSBlob(1234), mysql.MSBlob()),
+ (mysql.MSMediumBlob(),),
+ (mysql.MSLongBlob(),),
+ (mysql.ENUM("''","'fleem'"), ),
]
columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
from sqlalchemy import *
from sqlalchemy import sql, exc, schema
from sqlalchemy.util import u
+from sqlalchemy import util
from sqlalchemy.dialects.mysql import base as mysql
from sqlalchemy.testing import fixtures, AssertsCompiledSQL, AssertsExecutionResults
from sqlalchemy import testing
@testing.only_if('mysql')
@testing.exclude('mysql', '<', (5, 0, 5), 'a 5.0+ feature')
- @testing.fails_on('mysql+oursql', 'some round trips fail, oursql bug ?')
+ @testing.fails_if(
+ lambda: testing.against("mysql+oursql") and util.py3k,
+ 'some round trips fail, oursql bug ?')
@testing.provide_metadata
def test_bit_50_roundtrip(self):
bit_table = Table('mysql_bits', self.metadata,
self.assert_(colspec(table.c.y1).startswith('y1 YEAR'))
eq_(colspec(table.c.y5), 'y5 YEAR(4)')
- @testing.only_if('mysql')
- @testing.provide_metadata
- def test_set(self):
- """Exercise the SET type."""
-
- set_table = Table('mysql_set', self.metadata,
- Column('s1',
- mysql.MSSet("'dq'", "'sq'")), Column('s2',
- mysql.MSSet("'a'")), Column('s3',
- mysql.MSSet("'5'", "'7'", "'9'")))
- eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
- eq_(colspec(set_table.c.s2), "s2 SET('a')")
- eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
- set_table.create()
- reflected = Table('mysql_set', MetaData(testing.db),
- autoload=True)
- for table in set_table, reflected:
-
- def roundtrip(store, expected=None):
- expected = expected or store
- table.insert(store).execute()
- row = table.select().execute().first()
- self.assert_(list(row) == expected)
- table.delete().execute()
-
- roundtrip([None, None, None], [None] * 3)
- roundtrip(['', '', ''], [set([''])] * 3)
- roundtrip([set(['dq']), set(['a']), set(['5'])])
- roundtrip(['dq', 'a', '5'], [set(['dq']), set(['a']),
- set(['5'])])
- roundtrip([1, 1, 1], [set(['dq']), set(['a']), set(['5'
- ])])
- roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'
- ])])
- set_table.insert().execute({'s3': set(['5'])},
- {'s3': set(['5', '7'])}, {'s3': set(['5', '7', '9'])},
- {'s3': set(['7', '9'])})
-
- # NOTE: the string sent to MySQL here is sensitive to ordering.
- # for some reason the set ordering is always "5, 7" when we test on
- # MySQLdb but in Py3K this is not guaranteed. So basically our
- # SET type doesn't do ordering correctly (not sure how it can,
- # as we don't know how the SET was configured in the first place.)
- rows = select([set_table.c.s3],
- set_table.c.s3.in_([set(['5']), ['5', '7']])
- ).execute().fetchall()
- found = set([frozenset(row[0]) for row in rows])
- eq_(found, set([frozenset(['5']), frozenset(['5', '7'])]))
-class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
+class EnumSetTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
__only_on__ = 'mysql'
__dialect__ = mysql.dialect()
- @testing.uses_deprecated('Manually quoting ENUM value literals')
@testing.provide_metadata
def test_enum(self):
"""Exercise the ENUM type."""
+ with testing.expect_deprecated('Manually quoting ENUM value literals'):
+ e1, e2 = mysql.ENUM("'a'", "'b'"), mysql.ENUM("'a'", "'b'")
+
enum_table = Table('mysql_enum', self.metadata,
- Column('e1', mysql.ENUM("'a'", "'b'")),
- Column('e2', mysql.ENUM("'a'", "'b'"),
- nullable=False),
- Column('e2generic', Enum("a", "b"),
- nullable=False),
+ Column('e1', e1),
+ Column('e2', e2, nullable=False),
+ Column('e2generic', Enum("a", "b"), nullable=False),
Column('e3', mysql.ENUM("'a'", "'b'", strict=True)),
Column('e4', mysql.ENUM("'a'", "'b'", strict=True),
nullable=False),
eq_(res, expected)
+ @testing.provide_metadata
+ def test_set(self):
+
+ with testing.expect_deprecated('Manually quoting SET value literals'):
+ e1, e2 = mysql.SET("'a'", "'b'"), mysql.SET("'a'", "'b'")
+
+ set_table = Table('mysql_set', self.metadata,
+ Column('e1', e1),
+ Column('e2', e2, nullable=False),
+ Column('e3', mysql.SET("a", "b")),
+ Column('e4', mysql.SET("'a'", "b")),
+ Column('e5', mysql.SET("'a'", "'b'", quoting="quoted"))
+ )
+
+ eq_(colspec(set_table.c.e1),
+ "e1 SET('a','b')")
+ eq_(colspec(set_table.c.e2),
+ "e2 SET('a','b') NOT NULL")
+ eq_(colspec(set_table.c.e3),
+ "e3 SET('a','b')")
+ eq_(colspec(set_table.c.e4),
+ "e4 SET('''a''','b')")
+ eq_(colspec(set_table.c.e5),
+ "e5 SET('a','b')")
+ set_table.create()
+
+ assert_raises(exc.DBAPIError, set_table.insert().execute,
+ e1=None, e2=None, e3=None, e4=None)
+
+ if testing.against("+oursql"):
+ assert_raises(exc.StatementError, set_table.insert().execute,
+ e1='c', e2='c', e3='c', e4='c')
+
+ set_table.insert().execute(e1='a', e2='a', e3='a', e4="'a'", e5="a,b")
+ set_table.insert().execute(e1='b', e2='b', e3='b', e4='b', e5="a,b")
+
+ res = set_table.select().execute().fetchall()
+
+ if testing.against("+oursql"):
+ expected = [
+ # 1st row with all c's, data truncated
+ (set(['']), set(['']), set(['']), set(['']), None),
+ ]
+ else:
+ expected = []
+
+ expected.extend([
+ (set(['a']), set(['a']), set(['a']), set(["'a'"]), set(['a', 'b'])),
+ (set(['b']), set(['b']), set(['b']), set(['b']), set(['a', 'b']))
+ ])
+
+ eq_(res, expected)
+
+ @testing.provide_metadata
+ def test_set_roundtrip_plus_reflection(self):
+ set_table = Table('mysql_set', self.metadata,
+ Column('s1',
+ mysql.SET("dq", "sq")),
+ Column('s2', mysql.SET("a")),
+ Column('s3', mysql.SET("5", "7", "9")))
+
+ eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
+ eq_(colspec(set_table.c.s2), "s2 SET('a')")
+ eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
+ set_table.create()
+ reflected = Table('mysql_set', MetaData(testing.db),
+ autoload=True)
+ for table in set_table, reflected:
+
+ def roundtrip(store, expected=None):
+ expected = expected or store
+ table.insert(store).execute()
+ row = table.select().execute().first()
+ self.assert_(list(row) == expected)
+ table.delete().execute()
+
+ roundtrip([None, None, None], [None] * 3)
+ roundtrip(['', '', ''], [set([''])] * 3)
+ roundtrip([set(['dq']), set(['a']), set(['5'])])
+ roundtrip(['dq', 'a', '5'], [set(['dq']), set(['a']),
+ set(['5'])])
+ roundtrip([1, 1, 1], [set(['dq']), set(['a']), set(['5'
+ ])])
+ roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'
+ ])])
+ set_table.insert().execute({'s3': set(['5'])},
+ {'s3': set(['5', '7'])}, {'s3': set(['5', '7', '9'])},
+ {'s3': set(['7', '9'])})
+
+ # NOTE: the string sent to MySQL here is sensitive to ordering.
+ # for some reason the set ordering is always "5, 7" when we test on
+ # MySQLdb but in Py3K this is not guaranteed. So basically our
+ # SET type doesn't do ordering correctly (not sure how it can,
+ # as we don't know how the SET was configured in the first place.)
+ rows = select([set_table.c.s3],
+ set_table.c.s3.in_([set(['5']), ['5', '7']])
+ ).execute().fetchall()
+ found = set([frozenset(row[0]) for row in rows])
+ eq_(found, set([frozenset(['5']), frozenset(['5', '7'])]))
+
def test_unicode_enum(self):
unicode_engine = utf8_engine()
metadata = MetaData(unicode_engine)
"VARCHAR(1), CHECK (somecolumn IN ('x', "
"'y', 'z')))")
+ @testing.provide_metadata
@testing.exclude('mysql', '<', (4,), "3.23 can't handle an ENUM of ''")
- @testing.uses_deprecated('Manually quoting ENUM value literals')
def test_enum_parse(self):
- """More exercises for the ENUM type."""
-
- # MySQL 3.23 can't handle an ENUM of ''....
- enum_table = Table('mysql_enum', MetaData(testing.db),
- Column('e1', mysql.ENUM("'a'")),
- Column('e2', mysql.ENUM("''")),
- Column('e3', mysql.ENUM('a')),
- Column('e4', mysql.ENUM('')),
- Column('e5', mysql.ENUM("'a'", "''")),
- Column('e6', mysql.ENUM("''", "'a'")),
- Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
+ with testing.expect_deprecated('Manually quoting ENUM value literals'):
+ enum_table = Table('mysql_enum', self.metadata,
+ Column('e1', mysql.ENUM("'a'")),
+ Column('e2', mysql.ENUM("''")),
+ Column('e3', mysql.ENUM('a')),
+ Column('e4', mysql.ENUM('')),
+ Column('e5', mysql.ENUM("'a'", "''")),
+ Column('e6', mysql.ENUM("''", "'a'")),
+ Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
for col in enum_table.c:
self.assert_(repr(col))
- try:
- enum_table.create()
- reflected = Table('mysql_enum', MetaData(testing.db),
- autoload=True)
- for t in enum_table, reflected:
- eq_(t.c.e1.type.enums, ("a",))
- eq_(t.c.e2.type.enums, ("",))
- eq_(t.c.e3.type.enums, ("a",))
- eq_(t.c.e4.type.enums, ("",))
- eq_(t.c.e5.type.enums, ("a", ""))
- eq_(t.c.e6.type.enums, ("", "a"))
- eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
- finally:
- enum_table.drop()
+
+ enum_table.create()
+ reflected = Table('mysql_enum', MetaData(testing.db),
+ autoload=True)
+ for t in enum_table, reflected:
+ eq_(t.c.e1.type.enums, ("a",))
+ eq_(t.c.e2.type.enums, ("",))
+ eq_(t.c.e3.type.enums, ("a",))
+ eq_(t.c.e4.type.enums, ("",))
+ eq_(t.c.e5.type.enums, ("a", ""))
+ eq_(t.c.e6.type.enums, ("", "a"))
+ eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
+
+ @testing.provide_metadata
+ @testing.exclude('mysql', '<', (5,))
+ def test_set_parse(self):
+ with testing.expect_deprecated('Manually quoting SET value literals'):
+ set_table = Table('mysql_set', self.metadata,
+ Column('e1', mysql.SET("'a'")),
+ Column('e2', mysql.SET("''")),
+ Column('e3', mysql.SET('a')),
+ Column('e4', mysql.SET('')),
+ Column('e5', mysql.SET("'a'", "''")),
+ Column('e6', mysql.SET("''", "'a'")),
+ Column('e7', mysql.SET("''", "'''a'''", "'b''b'", "''''")))
+
+ for col in set_table.c:
+ self.assert_(repr(col))
+
+ set_table.create()
+
+ # don't want any warnings on reflection
+ reflected = Table('mysql_set', MetaData(testing.db),
+ autoload=True)
+ for t in set_table, reflected:
+ eq_(t.c.e1.type.values, ("a",))
+ eq_(t.c.e2.type.values, ("",))
+ eq_(t.c.e3.type.values, ("a",))
+ eq_(t.c.e4.type.values, ("",))
+ eq_(t.c.e5.type.values, ("a", ""))
+ eq_(t.c.e6.type.values, ("", "a"))
+ eq_(t.c.e7.type.values, ("", "'a'", "b'b", "'"))
def colspec(c):
return testing.db.dialect.ddl_compiler(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
