"""Support for the MySQL database.
SQLAlchemy supports 6 major MySQL versions: 3.23, 4.0, 4.1, 5.0, 5.1 and 6.0,
-with capablities increasing with more modern servers.
+with capablities increasing with more modern servers.
Versions 4.1 and higher support the basic SQL functionality that SQLAlchemy
uses in the ORM and SQL expressions. These versions pass the applicable
... Column('mytable_id', Integer, primary_key=True))
>>> t.create()
CREATE TABLE mytable (
- id INTEGER NOT NULL AUTO_INCREMENT,
+ id INTEGER NOT NULL AUTO_INCREMENT,
PRIMARY KEY (id)
)
__all__ = (
- 'MSBigInteger', 'MSBinary', 'MSBit', 'MSBlob', 'MSBoolean',
- 'MSChar', 'MSDate', 'MSDateTime', 'MSDecimal', 'MSDouble',
- 'MSEnum', 'MSFloat', 'MSInteger', 'MSLongBlob', 'MSLongText',
- 'MSMediumBlob', 'MSMediumText', 'MSNChar', 'MSNVarChar',
- 'MSNumeric', 'MSSet', 'MSSmallInteger', 'MSString', 'MSText',
- 'MSTime', 'MSTimeStamp', 'MSTinyBlob', 'MSTinyInteger',
+ 'MSBigInteger', 'MSBinary', 'MSBit', 'MSBlob', 'MSBoolean',
+ 'MSChar', 'MSDate', 'MSDateTime', 'MSDecimal', 'MSDouble',
+ 'MSEnum', 'MSFloat', 'MSInteger', 'MSLongBlob', 'MSLongText',
+ 'MSMediumBlob', 'MSMediumText', 'MSNChar', 'MSNVarChar',
+ 'MSNumeric', 'MSSet', 'MSSmallInteger', 'MSString', 'MSText',
+ 'MSTime', 'MSTimeStamp', 'MSTinyBlob', 'MSTinyInteger',
'MSTinyText', 'MSVarBinary', 'MSYear' )
def _extend(self, spec):
"Extend a numeric-type declaration with MySQL specific extensions."
-
+
if self.unsigned:
spec += ' UNSIGNED'
if self.zerofill:
ascii=False, unicode=False, binary=False,
national=False, **kwargs):
self.charset = charset
- # allow collate= or collation=
+ # allow collate= or collation=
self.collation = kwargs.get('collate', collation)
self.ascii = ascii
self.unicode = unicode
"""Extend a string-type declaration with standard SQL CHARACTER SET /
COLLATE annotations and MySQL specific extensions.
"""
-
+
if self.charset:
charset = 'CHARACTER SET %s' % self.charset
elif self.ascii:
collation = 'BINARY'
else:
collation = None
-
+
if self.national:
# NATIONAL (aka NCHAR/NVARCHAR) trumps charsets.
return ' '.join([c for c in ('NATIONAL', spec, collation)
def __repr__(self):
attributes = inspect.getargspec(self.__init__)[0][1:]
attributes.extend(inspect.getargspec(_StringType.__init__)[0][1:])
-
+
params = {}
for attr in attributes:
val = getattr(self, attr)
class MSNumeric(sqltypes.Numeric, _NumericType):
"""MySQL NUMERIC type."""
-
+
def __init__(self, precision=10, length=2, asdecimal=True, **kw):
"""Construct a NUMERIC.
_NumericType.__init__(self, **kw)
sqltypes.Numeric.__init__(self, precision, length, asdecimal=asdecimal)
-
+
def get_col_spec(self):
if self.precision is None:
return self._extend("NUMERIC")
"""
super(MSDecimal, self).__init__(precision, length, asdecimal=asdecimal, **kw)
-
+
def get_col_spec(self):
if self.precision is None:
return self._extend("DECIMAL")
def __init__(self, precision=10, length=None, asdecimal=False, **kw):
"""Construct a FLOAT.
-
+
precision
Total digits in this number. If length and precision are both None,
values are stored to limits allowed by the server.
This type is for MySQL 5.0.3 or greater for MyISAM, and 5.0.5 or greater for
MyISAM, MEMORY, InnoDB and BDB. For older versions, use a MSTinyInteger(1)
- type.
+ type.
"""
-
+
def __init__(self, length=None):
self.length = length
-
+
def result_processor(self, dialect):
"""Convert a MySQL's 64 bit, variable length binary string to a long."""
def process(value):
value = v
return value
return process
-
+
def get_col_spec(self):
if self.length is not None:
return "BIT(%s)" % self.length
return "YEAR(%s)" % self.length
class MSText(_StringType, sqltypes.TEXT):
- """MySQL TEXT type, for text up to 2^16 characters."""
-
+ """MySQL TEXT type, for text up to 2^16 characters."""
+
def __init__(self, length=None, **kwargs):
"""Construct a TEXT.
-
+
length
Optional, if provided the server may optimize storage by
subsitituting the smallest TEXT type sufficient to store
return self._extend("TEXT(%d)" % self.length)
else:
return self._extend("TEXT")
-
+
class MSTinyText(MSText):
- """MySQL TINYTEXT type, for text up to 2^8 characters."""
+ """MySQL TINYTEXT type, for text up to 2^8 characters."""
def __init__(self, **kwargs):
"""Construct a TINYTEXT.
-
+
charset
Optional, a column-level character set for this string
value. Takes precendence to 'ascii' or 'unicode' short-hand.
class MSMediumText(MSText):
- """MySQL MEDIUMTEXT type, for text up to 2^24 characters."""
+ """MySQL MEDIUMTEXT type, for text up to 2^24 characters."""
def __init__(self, **kwargs):
"""Construct a MEDIUMTEXT.
-
+
charset
Optional, a column-level character set for this string
value. Takes precendence to 'ascii' or 'unicode' short-hand.
class MSLongText(MSText):
- """MySQL LONGTEXT type, for text up to 2^32 characters."""
+ """MySQL LONGTEXT type, for text up to 2^32 characters."""
def __init__(self, **kwargs):
"""Construct a LONGTEXT.
-
+
charset
Optional, a column-level character set for this string
value. Takes precendence to 'ascii' or 'unicode' short-hand.
def __init__(self, length=None, **kwargs):
"""Construct a VARCHAR.
-
+
length
Maximum data length, in characters.
class MSChar(_StringType, sqltypes.CHAR):
"""MySQL CHAR type, for fixed-length character data."""
-
+
def __init__(self, length, **kwargs):
"""Construct an NCHAR.
-
+
length
Maximum data length, in characters.
def __init__(self, length=None, **kwargs):
"""Construct an NVARCHAR.
-
+
length
Maximum data length, in characters.
# We'll actually generate the equiv. "NATIONAL VARCHAR" instead
# of "NVARCHAR".
return self._extend("VARCHAR(%(length)s)" % {'length': self.length})
-
+
class MSNChar(_StringType, sqltypes.CHAR):
"""MySQL NCHAR type.
return process
class MSBlob(_BinaryType):
- """MySQL BLOB type, for binary data up to 2^16 bytes"""
+ """MySQL BLOB type, for binary data up to 2^16 bytes"""
def __init__(self, length=None, **kw):
"""Construct a BLOB. Arguments are:
return "BLOB(%d)" % self.length
else:
return "BLOB"
-
+
def result_processor(self, dialect):
def process(value):
if value is None:
else:
return buffer(value)
return process
-
+
def __repr__(self):
return "%s()" % self.__class__.__name__
class MSTinyBlob(MSBlob):
- """MySQL TINYBLOB type, for binary data up to 2^8 bytes."""
+ """MySQL TINYBLOB type, for binary data up to 2^8 bytes."""
def get_col_spec(self):
return "TINYBLOB"
-class MSMediumBlob(MSBlob):
+class MSMediumBlob(MSBlob):
"""MySQL MEDIUMBLOB type, for binary data up to 2^24 bytes."""
def get_col_spec(self):
class MSEnum(MSString):
"""MySQL ENUM type."""
-
+
def __init__(self, *enums, **kw):
- """
- Construct an ENUM.
+ """Construct an ENUM.
Example:
Column('myenum', MSEnum("'foo'", "'bar'", "'baz'"))
Arguments are:
-
+
enums
The range of valid values for this ENUM. Values will be used
exactly as they appear when generating schemas. Strings must
schema. This does not affect the type of data stored, only the
collation of character data.
"""
-
+
self.__ddl_values = enums
strip_enums = []
# strip enclosing quotes and unquote interior
a = a[1:-1].replace(a[0] * 2, a[0])
strip_enums.append(a)
-
+
self.enums = strip_enums
self.strict = kw.pop('strict', False)
length = max([len(v) for v in strip_enums] + [0])
else:
return value
return process
-
+
def get_col_spec(self):
return self._extend("ENUM(%s)" % ",".join(self.__ddl_values))
class MSSet(MSString):
"""MySQL SET type."""
-
+
def __init__(self, *values, **kw):
"""Construct a SET.
Column('myset', MSSet("'foo'", "'bar'", "'baz'"))
Arguments are:
-
+
values
The range of valid values for this SET. Values will be used
exactly as they appear when generating schemas. Strings must
schema. This does not affect the type of data stored, only the
collation of character data.
"""
-
+
self.__ddl_values = values
strip_values = []
# strip enclosing quotes and unquote interior
a = a[1:-1].replace(a[0] * 2, a[0])
strip_values.append(a)
-
+
self.values = strip_values
length = max([len(v) for v in strip_values] + [0])
super(MSSet, self).__init__(length, **kw)
else:
return value
return process
-
+
def bind_processor(self, dialect):
super_convert = super(MSSet, self).bind_processor(dialect)
def process(value):
else:
return value
return process
-
+
def get_col_spec(self):
return self._extend("SET(%s)" % ",".join(self.__ddl_values))
return None
return value and True or False
return process
-
+
def bind_processor(self, dialect):
def process(value):
if value is True:
'bigint': MSBigInteger,
'binary': MSBinary,
'bit': MSBit,
- 'blob': MSBlob,
+ 'blob': MSBlob,
'boolean':MSBoolean,
'char': MSChar,
'date': MSDate,
self._last_inserted_ids[0] is None):
self._last_inserted_ids = ([self.cursor.lastrowid] +
self._last_inserted_ids[1:])
-
+
def is_select(self):
return SELECT_RE.match(self.statement)
import MySQLdb as mysql
return mysql
dbapi = classmethod(dbapi)
-
+
def create_connect_args(self, url):
opts = url.translate_connect_args(database='db', username='user',
password='passwd')
# Note: using either of the below will cause all strings to be returned
# as Unicode, both in raw SQL operations and with column types like
# String and MSString.
- util.coerce_kw_type(opts, 'use_unicode', bool)
+ util.coerce_kw_type(opts, 'use_unicode', bool)
util.coerce_kw_type(opts, 'charset', str)
# Rich values 'cursorclass' and 'conv' are not supported via
# query string.
-
+
ssl = {}
for key in ['ssl_ca', 'ssl_key', 'ssl_cert', 'ssl_capath', 'ssl_cipher']:
if key in opts:
del opts[key]
if ssl:
opts['ssl'] = ssl
-
+
# FOUND_ROWS must be set in CLIENT_FLAGS to enable
# supports_sane_rowcount.
client_flag = opts.get('client_flag', 0)
rowcount = cursor.executemany(statement, parameters)
if context is not None:
context._rowcount = rowcount
-
+
def supports_unicode_statements(self):
return True
-
+
def do_execute(self, cursor, statement, parameters, context=None):
cursor.execute(statement, parameters)
def do_rollback(self, connection):
"""Execute a ROLLBACK."""
-
+
try:
connection.rollback()
except:
if not is_prepared:
self.do_prepare_twophase(connection, xid)
connection.execute("XA COMMIT %s", xid)
-
+
def do_recover_twophase(self, connection):
resultset = connection.execute("XA RECOVER")
return [row['data'][0:row['gtrid_length']] for row in resultset]
full_name = '.'.join(self.identifier_preparer._quote_free_identifiers(
schema, table_name))
-
+
st = "DESCRIBE %s" % full_name
rs = None
try:
self.reflector = reflector = \
MySQLSchemaReflector(self.identifier_preparer)
- sql = self._show_create_table(connection, table, charset)
+ sql = self._show_create_table(connection, table, charset)
if sql.startswith('CREATE ALGORITHM'):
# Adapt views to something table-like.
columns = self._describe_table(connection, table, charset)
def _adjust_casing(self, connection, table, charset=None):
"""Adjust Table name to the server case sensitivity, if needed."""
-
+
if charset is None:
charset = self._detect_charset(connection)
-
+
casing = self._detect_casing(connection, charset)
# For winxx database hosts. TODO: is this really needed?
table.name = table.name.lower()
lc_alias = schema._get_table_key(table.name, table.schema)
table.metadata.tables[lc_alias] = table
-
+
def _detect_charset(self, connection):
"""Sniff out the character set in use for connection results."""
# Prefer 'character_set_results' for the current connection over the
# value in the driver. SET NAMES or individual variable SETs will
# change the charset without updating the driver's view of the world.
- #
+ #
# If it's decided that issuing that sort of SQL leaves you SOL, then
# this can prefer the driver value.
rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'")
Cached per-connection.
"""
-
+
try:
return connection.properties['collations']
except KeyError:
def visit_update(self, update_stmt):
self.stack.append({'from':util.Set([update_stmt.table])})
-
+
self.isupdate = True
colparams = self._get_colparams(update_stmt)
limit = update_stmt.kwargs.get('mysql_limit', None)
if limit:
text += " LIMIT %s" % limit
-
+
self.stack.pop(-1)
-
+
return text
# ug. "InnoDB needs indexes on foreign keys and referenced keys [...].
class MySQLSchemaGenerator(compiler.SchemaGenerator):
def get_column_specification(self, column, first_pk=False):
"""Builds column DDL."""
-
+
colspec = [self.preparer.format_column(column),
column.type.dialect_impl(self.dialect).get_col_spec()]
if opt in ('TABLESPACE', 'DEFAULT CHARACTER SET',
'CHARACTER SET', 'COLLATE'):
joiner = ' '
-
+
table_opts.append(joiner.join((opt, table.kwargs[k])))
return ' '.join(table_opts)
An ANSIIdentifierPreparer type, used to determine the identifier
quoting style in effect.
"""
-
+
self.preparer = identifier_preparer
self._prep_regexes()
# No match.
return (None, line)
-
+
def parse_table_options(self, line):
"""Build a dictionary of all reflected table-level options.
line
The final line of SHOW CREATE TABLE output.
"""
-
+
options = {}
if not line or line == ')':
"""MySQL-specific schema identifier configuration."""
reserved_words = RESERVED_WORDS
-
+
def __init__(self, dialect, **kw):
super(_MySQLIdentifierPreparer, self).__init__(dialect, **kw)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
