from sqlalchemy.engine import default, base, reflection
from sqlalchemy import types as sqltypes
from decimal import Decimal as _python_Decimal
+from sqlalchemy.types import INTEGER, BIGINT, SMALLINT, DECIMAL, NUMERIC, \
+ FLOAT, TIMESTAMP
+
from sqlalchemy.dialects.mssql import information_schema as ischema
return process
-class MSReal(sqltypes.Float):
+class REAL(sqltypes.Float):
"""A type for ``real`` numbers."""
__visit_name__ = 'REAL'
def __init__(self):
- super(MSReal, self).__init__(precision=24)
+ super(REAL, self).__init__(precision=24)
+MSReal = REAL
-class MSTinyInteger(sqltypes.Integer):
+class TINYINT(sqltypes.Integer):
__visit_name__ = 'TINYINT'
+MSTinyInteger = TINYINT
+
# MSSQL DATE/TIME types have varied behavior, sometimes returning
# strings. MSDate/MSTime check for everything, and always
# filter bind parameters into datetime objects (required by pyodbc,
# not sure about other dialects).
-class MSDate(sqltypes.Date):
+class DATE(sqltypes.DATE):
def bind_processor(self, dialect):
def process(value):
if type(value) == datetime.date:
else:
return value
return process
-
-class MSTime(sqltypes.Time):
+MSDate = DATE
+
+class TIME(sqltypes.TIME):
def __init__(self, precision=None, **kwargs):
self.precision = precision
- super(MSTime, self).__init__()
+ super(TIME, self).__init__()
__zero_date = datetime.date(1900, 1, 1)
return value
return process
+MSTime = TIME
+
class _DateTimeBase(object):
def bind_processor(self, dialect):
def process(value):
+ # TODO: why ?
if type(value) == datetime.date:
return datetime.datetime(value.year, value.month, value.day)
else:
return value
return process
-class MSDateTime(_DateTimeBase, sqltypes.DATETIME):
+class DATETIME(_DateTimeBase, sqltypes.DATETIME):
pass
-
-class MSSmallDateTime(_DateTimeBase, sqltypes.DateTime):
+MSDateTime = DATETIME
+
+class SMALLDATETIME(_DateTimeBase, sqltypes.DateTime):
__visit_name__ = 'SMALLDATETIME'
+MSSmallDateTime = SMALLDATETIME
-class MSDateTime2(_DateTimeBase, sqltypes.DateTime):
+class DATETIME2(_DateTimeBase, sqltypes.DateTime):
__visit_name__ = 'DATETIME2'
def __init__(self, precision=None, **kwargs):
self.precision = precision
+MSDateTime2 = DATETIME2
-class MSDateTimeOffset(sqltypes.TypeEngine):
+
+# TODO: is this not an Interval ?
+class DATETIMEOFFSET(sqltypes.TypeEngine):
__visit_name__ = 'DATETIMEOFFSET'
def __init__(self, precision=None, **kwargs):
self.precision = precision
+MSDateTimeOffset = DATETIMEOFFSET
+
class _StringType(object):
"""Base for MSSQL string types."""
', '.join(['%s=%r' % (k, params[k]) for k in params]))
-class MSText(_StringType, sqltypes.TEXT):
+class TEXT(_StringType, sqltypes.TEXT):
"""MSSQL TEXT type, for variable-length text up to 2^31 characters."""
def __init__(self, *args, **kw):
collation = kw.pop('collation', None)
_StringType.__init__(self, collation)
sqltypes.Text.__init__(self, *args, **kw)
+MSText = TEXT
-class MSNText(_StringType, sqltypes.UnicodeText):
+class NTEXT(_StringType, sqltypes.UnicodeText):
"""MSSQL NTEXT type, for variable-length unicode text up to 2^30
characters."""
_StringType.__init__(self, collation)
length = kwargs.pop('length', None)
sqltypes.UnicodeText.__init__(self, length, **kwargs)
+MSNText = NTEXT
-
-class MSString(_StringType, sqltypes.VARCHAR):
+class VARCHAR(_StringType, sqltypes.VARCHAR):
"""MSSQL VARCHAR type, for variable-length non-Unicode data with a maximum
of 8,000 characters."""
collation = kw.pop('collation', None)
_StringType.__init__(self, collation)
sqltypes.VARCHAR.__init__(self, *args, **kw)
+MSString = VARCHAR
-class MSNVarchar(_StringType, sqltypes.NVARCHAR):
+class NVARCHAR(_StringType, sqltypes.NVARCHAR):
"""MSSQL NVARCHAR type.
For variable-length unicode character data up to 4,000 characters."""
collation = kw.pop('collation', None)
_StringType.__init__(self, collation)
sqltypes.NVARCHAR.__init__(self, *args, **kw)
+MSNVarchar = NVARCHAR
-
-class MSChar(_StringType, sqltypes.CHAR):
+class CHAR(_StringType, sqltypes.CHAR):
"""MSSQL CHAR type, for fixed-length non-Unicode data with a maximum
of 8,000 characters."""
collation = kw.pop('collation', None)
_StringType.__init__(self, collation)
sqltypes.CHAR.__init__(self, *args, **kw)
+MSChar = CHAR
-
-class MSNChar(_StringType, sqltypes.NCHAR):
+class NCHAR(_StringType, sqltypes.NCHAR):
"""MSSQL NCHAR type.
For fixed-length unicode character data up to 4,000 characters."""
collation = kw.pop('collation', None)
_StringType.__init__(self, collation)
sqltypes.NCHAR.__init__(self, *args, **kw)
+MSNChar = NCHAR
-class MSBinary(sqltypes.Binary):
+class BINARY(sqltypes.Binary):
__visit_name__ = 'BINARY'
+MSBinary = BINARY
-class MSVarBinary(sqltypes.Binary):
+class VARBINARY(sqltypes.Binary):
__visit_name__ = 'VARBINARY'
-
+MSVarBinary = VARBINARY
-class MSImage(sqltypes.Binary):
+class IMAGE(sqltypes.Binary):
__visit_name__ = 'IMAGE'
+MSImage = IMAGE
-class MSBit(sqltypes.TypeEngine):
+class BIT(sqltypes.TypeEngine):
__visit_name__ = 'BIT'
+MSBit = BIT
class MSBoolean(sqltypes.Boolean):
def result_processor(self, dialect):
return value and True or False
return process
-class MSMoney(sqltypes.TypeEngine):
+class MONEY(sqltypes.TypeEngine):
__visit_name__ = 'MONEY'
+MSMoney = MONEY
-class MSSmallMoney(MSMoney):
+class SMALLMONEY(MSMoney):
__visit_name__ = 'SMALLMONEY'
+MSSmallMoney = SMALLMONEY
-
-class MSUniqueIdentifier(sqltypes.TypeEngine):
+class UNIQUEIDENTIFIER(sqltypes.TypeEngine):
__visit_name__ = "UNIQUEIDENTIFIER"
+MSUniqueIdentifier = UNIQUEIDENTIFIER
-class MSVariant(sqltypes.TypeEngine):
+class SQL_VARIANT(sqltypes.TypeEngine):
__visit_name__ = 'SQL_VARIANT'
+MSVariant = SQL_VARIANT
class MSTypeCompiler(compiler.GenericTypeCompiler):
def _extend(self, spec, type_):
colspecs = {
sqltypes.Numeric : MSNumeric,
- sqltypes.DateTime : MSDateTime,
- sqltypes.Date : MSDate,
- sqltypes.Time : MSTime,
+ sqltypes.DateTime : DATETIME,
+ sqltypes.Date : DATE,
+ sqltypes.Time : TIME,
sqltypes.Boolean : MSBoolean,
}
ischema_names = {
- 'int' : sqltypes.INTEGER,
- 'bigint': sqltypes.BigInteger,
- 'smallint' : sqltypes.SmallInteger,
- 'tinyint' : MSTinyInteger,
- 'varchar' : MSString,
- 'nvarchar' : MSNVarchar,
- 'char' : MSChar,
- 'nchar' : MSNChar,
- 'text' : MSText,
- 'ntext' : MSNText,
- 'decimal' : sqltypes.DECIMAL,
- 'numeric' : sqltypes.NUMERIC,
- 'float' : sqltypes.FLOAT,
- 'datetime' : sqltypes.DATETIME,
- 'datetime2' : MSDateTime2,
- 'datetimeoffset' : MSDateTimeOffset,
- 'date': sqltypes.DATE,
- 'time': MSTime,
- 'smalldatetime' : MSSmallDateTime,
- 'binary' : MSBinary,
- 'varbinary' : MSVarBinary,
- 'bit': sqltypes.Boolean,
- 'real' : MSReal,
- 'image' : MSImage,
- 'timestamp': sqltypes.TIMESTAMP,
- 'money': MSMoney,
- 'smallmoney': MSSmallMoney,
- 'uniqueidentifier': MSUniqueIdentifier,
- 'sql_variant': MSVariant,
+ 'int' : INTEGER,
+ 'bigint': BIGINT,
+ 'smallint' : SMALLINT,
+ 'tinyint' : TINYINT,
+ 'varchar' : VARCHAR,
+ 'nvarchar' : NVARCHAR,
+ 'char' : CHAR,
+ 'nchar' : NCHAR,
+ 'text' : TEXT,
+ 'ntext' : NTEXT,
+ 'decimal' : DECIMAL,
+ 'numeric' : NUMERIC,
+ 'float' : FLOAT,
+ 'datetime' : DATETIME,
+ 'datetime2' : DATETIME2,
+ 'datetimeoffset' : DATETIMEOFFSET,
+ 'date': DATE,
+ 'time': TIME,
+ 'smalldatetime' : SMALLDATETIME,
+ 'binary' : BINARY,
+ 'varbinary' : VARBINARY,
+ 'bit': BIT,
+ 'real' : REAL,
+ 'image' : IMAGE,
+ 'timestamp': TIMESTAMP,
+ 'money': MONEY,
+ 'smallmoney': SMALLMONEY,
+ 'uniqueidentifier': UNIQUEIDENTIFIER,
+ 'sql_variant': SQL_VARIANT,
}
class MSSQLCompiler(compiler.SQLCompiler):
Table('mytable', metadata,
Column('id', Integer, primary_key=True),
- Column('ittybittyblob', mysql.MSTinyBlob),
- Column('biggy', mysql.MSBigInteger(unsigned=True)))
+ Column('ittybittyblob', mysql.TINYBLOB),
+ Column('biggy', mysql.BIGINT(unsigned=True)))
All standard MySQL column types are supported. The OpenGIS types are
available for use via table reflection but have no special support or mapping
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy import types as sqltypes
-
-__all__ = (
- 'MSBigInteger', 'MSMediumInteger', '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' )
-
+from sqlalchemy.types import DATE, DATETIME
RESERVED_WORDS = set(
['accessible', 'add', 'all', 'alter', 'analyze','and', 'as', 'asc',
class _NumericType(object):
"""Base for MySQL numeric types."""
- def __init__(self, kw):
+ def __init__(self, **kw):
self.unsigned = kw.pop('unsigned', False)
self.zerofill = kw.pop('zerofill', False)
+ super(_NumericType, self).__init__(**kw)
+
+class _FloatType(_NumericType, sqltypes.Float):
+ def __init__(self, precision=None, scale=None, asdecimal=True, **kw):
+ if isinstance(self, (REAL, DOUBLE)) and \
+ (
+ (precision is None and scale is not None) or
+ (precision is not None and scale is None)
+ ):
+ raise exc.ArgumentError(
+ "You must specify both precision and scale or omit "
+ "both altogether.")
+
+ super(_FloatType, self).__init__(precision=precision, asdecimal=asdecimal, **kw)
+ self.scale = scale
+
+class _DecimalType(_NumericType):
+ def bind_processor(self, dialect):
+ return None
+
+ def result_processor(self, dialect):
+ # TODO: this behavior might by MySQLdb specific,
+ # i.e. that Decimals are returned by the DBAPI
+ if not self.asdecimal:
+ def process(value):
+ if isinstance(value, decimal.Decimal):
+ return float(value)
+ else:
+ return value
+ return process
+ else:
+ return None
+class _IntegerType(_NumericType, sqltypes.Integer):
+ def __init__(self, display_width=None, **kw):
+ self.display_width = display_width
+ super(_IntegerType, self).__init__(**kw)
-class _StringType(object):
+class _StringType(sqltypes.String):
"""Base for MySQL string types."""
def __init__(self, charset=None, collation=None,
ascii=False, unicode=False, binary=False,
- national=False, **kwargs):
+ national=False, **kw):
self.charset = charset
# allow collate= or collation=
- self.collation = kwargs.get('collate', collation)
+ self.collation = kw.pop('collate', collation)
self.ascii = ascii
self.unicode = unicode
self.binary = binary
self.national = national
-
+ super(_StringType, self).__init__(**kw)
+
def __repr__(self):
attributes = inspect.getargspec(self.__init__)[0][1:]
attributes.extend(inspect.getargspec(_StringType.__init__)[0][1:])
', '.join(['%s=%r' % (k, params[k]) for k in params]))
-class MSNumeric(sqltypes.Numeric, _NumericType):
+class _BinaryType(sqltypes.Binary):
+ """Base for MySQL binary types."""
+
+ def result_processor(self, dialect):
+ def process(value):
+ if value is None:
+ return None
+ else:
+ return util.buffer(value)
+ return process
+
+class NUMERIC(_DecimalType, sqltypes.NUMERIC):
"""MySQL NUMERIC type."""
__visit_name__ = 'NUMERIC'
numeric.
"""
- _NumericType.__init__(self, kw)
- sqltypes.Numeric.__init__(self, precision, scale, asdecimal=asdecimal, **kw)
+ super(NUMERIC, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal, **kw)
- def bind_processor(self, dialect):
- return None
- def result_processor(self, dialect):
- if not self.asdecimal:
- def process(value):
- if isinstance(value, decimal.Decimal):
- return float(value)
- else:
- return value
- return process
- else:
- return None
-
-
-class MSDecimal(MSNumeric):
+class DECIMAL(_DecimalType, sqltypes.DECIMAL):
"""MySQL DECIMAL type."""
__visit_name__ = 'DECIMAL'
numeric.
"""
- super(MSDecimal, self).__init__(precision, scale, asdecimal=asdecimal, **kw)
-
+ super(DECIMAL, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal, **kw)
-class MSDouble(sqltypes.Float, _NumericType):
+
+class DOUBLE(_FloatType):
"""MySQL DOUBLE type."""
__visit_name__ = 'DOUBLE'
numeric.
"""
- if ((precision is None and scale is not None) or
- (precision is not None and scale is None)):
- raise exc.ArgumentError(
- "You must specify both precision and scale or omit "
- "both altogether.")
-
- _NumericType.__init__(self, kw)
- sqltypes.Float.__init__(self, asdecimal=asdecimal, **kw)
- self.scale = scale
- self.precision = precision
+ super(DOUBLE, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal, **kw)
-
-class MSReal(MSDouble):
+class REAL(_FloatType):
"""MySQL REAL type."""
__visit_name__ = 'REAL'
numeric.
"""
- MSDouble.__init__(self, precision, scale, asdecimal, **kw)
-
+ super(REAL, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal, **kw)
-class MSFloat(sqltypes.Float, _NumericType):
+class FLOAT(_FloatType, sqltypes.FLOAT):
"""MySQL FLOAT type."""
__visit_name__ = 'FLOAT'
numeric.
"""
- _NumericType.__init__(self, kw)
- sqltypes.Float.__init__(self, asdecimal=asdecimal, **kw)
- self.scale = scale
- self.precision = precision
+ super(FLOAT, self).__init__(precision=precision, scale=scale, asdecimal=asdecimal, **kw)
def bind_processor(self, dialect):
return None
-
-class MSInteger(sqltypes.Integer, _NumericType):
+class INTEGER(_IntegerType, sqltypes.INTEGER):
"""MySQL INTEGER type."""
__visit_name__ = 'INTEGER'
numeric.
"""
- if 'length' in kw:
- util.warn_deprecated("'length' is deprecated for MSInteger and subclasses. Use 'display_width'.")
- self.display_width = kw.pop('length')
- else:
- self.display_width = display_width
- _NumericType.__init__(self, kw)
- sqltypes.Integer.__init__(self, **kw)
-
+ super(INTEGER, self).__init__(display_width=display_width, **kw)
-class MSBigInteger(MSInteger):
+class BIGINT(_IntegerType, sqltypes.BIGINT):
"""MySQL BIGINTEGER type."""
__visit_name__ = 'BIGINT'
numeric.
"""
- super(MSBigInteger, self).__init__(display_width, **kw)
+ super(BIGINT, self).__init__(display_width=display_width, **kw)
-
-class MSMediumInteger(MSInteger):
+class MEDIUMINT(_IntegerType):
"""MySQL MEDIUMINTEGER type."""
__visit_name__ = 'MEDIUMINT'
numeric.
"""
- super(MSMediumInteger, self).__init__(display_width, **kw)
-
+ super(MEDIUMINT, self).__init__(display_width=display_width, **kw)
-class MSTinyInteger(MSInteger):
+class TINYINT(_IntegerType):
"""MySQL TINYINT type."""
__visit_name__ = 'TINYINT'
numeric.
"""
- super(MSTinyInteger, self).__init__(display_width, **kw)
+ super(TINYINT, self).__init__(display_width=display_width, **kw)
-
-class MSSmallInteger(sqltypes.SmallInteger, MSInteger):
+class SMALLINT(_IntegerType, sqltypes.SMALLINT):
"""MySQL SMALLINTEGER type."""
__visit_name__ = 'SMALLINT'
numeric.
"""
- self.display_width = display_width
- _NumericType.__init__(self, kw)
- sqltypes.SmallInteger.__init__(self, **kw)
+ super(SMALLINT, self).__init__(display_width=display_width, **kw)
-
-class MSBit(sqltypes.TypeEngine):
+class BIT(sqltypes.TypeEngine):
"""MySQL BIT type.
This type is for MySQL 5.0.3 or greater for MyISAM, and 5.0.5 or greater for
return value
return process
-# TODO: probably don't need datetime/date types since no behavior changes
-
-class MSDateTime(sqltypes.DateTime):
- """MySQL DATETIME type."""
-
- __visit_name__ = 'DATETIME'
-
-
-class MSDate(sqltypes.Date):
- """MySQL DATE type."""
- __visit_name__ = 'DATE'
-
-
-
-class MSTime(sqltypes.Time):
+class TIME(sqltypes.TIME):
"""MySQL TIME type."""
__visit_name__ = 'TIME'
return None
return process
-class MSTimeStamp(sqltypes.TIMESTAMP):
- """MySQL TIMESTAMP type.
-
- To signal the orm to automatically re-select modified rows to retrieve the
- updated timestamp, add a ``server_default`` to your
- :class:`~sqlalchemy.Column` specification::
-
- from sqlalchemy.databases import mysql
- Column('updated', mysql.MSTimeStamp,
- server_default=sql.text('CURRENT_TIMESTAMP')
- )
-
- The full range of MySQL 4.1+ TIMESTAMP defaults can be specified in
- the the default::
-
- server_default=sql.text('CURRENT TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')
-
- """
+class TIMESTAMP(sqltypes.TIMESTAMP):
+ """MySQL TIMESTAMP type."""
__visit_name__ = 'TIMESTAMP'
-
-class MSYear(sqltypes.TypeEngine):
+class YEAR(sqltypes.TypeEngine):
"""MySQL YEAR type, for single byte storage of years 1901-2155."""
__visit_name__ = 'YEAR'
def __init__(self, display_width=None):
self.display_width = display_width
-
-class MSText(_StringType, sqltypes.Text):
+class TEXT(_StringType, sqltypes.TEXT):
"""MySQL TEXT type, for text up to 2^16 characters."""
__visit_name__ = 'TEXT'
- def __init__(self, length=None, **kwargs):
+ def __init__(self, length=None, **kw):
"""Construct a TEXT.
:param length: Optional, if provided the server may optimize storage
only the collation of character data.
"""
- _StringType.__init__(self, **kwargs)
- sqltypes.Text.__init__(self, length,
- kwargs.get('convert_unicode', False), kwargs.get('assert_unicode', None))
-
+ super(TEXT, self).__init__(length=length, **kw)
-class MSTinyText(MSText):
+class TINYTEXT(_StringType):
"""MySQL TINYTEXT type, for text up to 2^8 characters."""
__visit_name__ = 'TINYTEXT'
only the collation of character data.
"""
+ super(TINYTEXT, self).__init__(**kwargs)
- super(MSTinyText, self).__init__(**kwargs)
-
-
-class MSMediumText(MSText):
+class MEDIUMTEXT(_StringType):
"""MySQL MEDIUMTEXT type, for text up to 2^24 characters."""
__visit_name__ = 'MEDIUMTEXT'
only the collation of character data.
"""
- super(MSMediumText, self).__init__(**kwargs)
+ super(MEDIUMTEXT, self).__init__(**kwargs)
-class MSLongText(MSText):
+class LONGTEXT(_StringType):
"""MySQL LONGTEXT type, for text up to 2^32 characters."""
__visit_name__ = 'LONGTEXT'
only the collation of character data.
"""
- super(MSLongText, self).__init__(**kwargs)
-
-
+ super(LONGTEXT, self).__init__(**kwargs)
-class MSString(_StringType, sqltypes.String):
+
+class VARCHAR(_StringType, sqltypes.VARCHAR):
"""MySQL VARCHAR type, for variable-length character data."""
__visit_name__ = 'VARCHAR'
only the collation of character data.
"""
- _StringType.__init__(self, **kwargs)
- sqltypes.String.__init__(self, length,
- kwargs.get('convert_unicode', False), kwargs.get('assert_unicode', None))
-
+ super(VARCHAR, self).__init__(length=length, **kwargs)
-class MSChar(_StringType, sqltypes.CHAR):
+class CHAR(_StringType, sqltypes.CHAR):
"""MySQL CHAR type, for fixed-length character data."""
__visit_name__ = 'CHAR'
compatible with the national character set.
"""
- _StringType.__init__(self, **kwargs)
- sqltypes.CHAR.__init__(self, length,
- kwargs.get('convert_unicode', False))
-
-
+ super(CHAR, self).__init__(length=length, **kwargs)
-class MSNVarChar(_StringType, sqltypes.String):
+class NVARCHAR(_StringType, sqltypes.NVARCHAR):
"""MySQL NVARCHAR type.
For variable-length character data in the server's configured national
"""
kwargs['national'] = True
- _StringType.__init__(self, **kwargs)
- sqltypes.String.__init__(self, length,
- kwargs.get('convert_unicode', False))
+ super(NVARCHAR, self).__init__(length=length, **kwargs)
-
-class MSNChar(_StringType, sqltypes.CHAR):
+class NCHAR(_StringType, sqltypes.NCHAR):
"""MySQL NCHAR type.
For fixed-length character data in the server's configured national
"""
kwargs['national'] = True
- _StringType.__init__(self, **kwargs)
- sqltypes.CHAR.__init__(self, length,
- kwargs.get('convert_unicode', False))
-
+ super(NCHAR, self).__init__(length=length, **kwargs)
-class _BinaryType(sqltypes.Binary):
- """Base for MySQL binary types."""
- def result_processor(self, dialect):
- def process(value):
- if value is None:
- return None
- else:
- return util.buffer(value)
- return process
-class MSVarBinary(_BinaryType):
+class VARBINARY(_BinaryType):
"""MySQL VARBINARY type, for variable length binary data."""
__visit_name__ = 'VARBINARY'
:param length: Maximum data length, in characters.
"""
- super(MSVarBinary, self).__init__(length, **kw)
-
+ super(VARBINARY, self).__init__(length=length, **kw)
-class MSBinary(_BinaryType):
+class BINARY(_BinaryType):
"""MySQL BINARY type, for fixed length binary data"""
__visit_name__ = 'BINARY'
specified, this will generate a BLOB. This usage is deprecated.
"""
- super(MSBinary, self).__init__(length, **kw)
+ super(BINARY, self).__init__(length=length, **kw)
- def result_processor(self, dialect):
- def process(value):
- if value is None:
- return None
- else:
- return util.buffer(value)
- return process
-
-class MSBlob(_BinaryType):
+class BLOB(_BinaryType, sqltypes.BLOB):
"""MySQL BLOB type, for binary data up to 2^16 bytes"""
__visit_name__ = 'BLOB'
``length`` characters.
"""
- super(MSBlob, self).__init__(length, **kw)
+ super(BLOB, self).__init__(length=length, **kw)
- def result_processor(self, dialect):
- def process(value):
- if value is None:
- return None
- else:
- return util.buffer(value)
- return process
-
- def __repr__(self):
- return "%s()" % self.__class__.__name__
-
-class MSTinyBlob(MSBlob):
+class TINYBLOB(_BinaryType):
"""MySQL TINYBLOB type, for binary data up to 2^8 bytes."""
__visit_name__ = 'TINYBLOB'
-
-class MSMediumBlob(MSBlob):
+class MEDIUMBLOB(_BinaryType):
"""MySQL MEDIUMBLOB type, for binary data up to 2^24 bytes."""
__visit_name__ = 'MEDIUMBLOB'
-
-class MSLongBlob(MSBlob):
+class LONGBLOB(_BinaryType):
"""MySQL LONGBLOB type, for binary data up to 2^32 bytes."""
__visit_name__ = 'LONGBLOB'
-
-class MSEnum(MSString):
+class ENUM(_StringType):
"""MySQL ENUM type."""
__visit_name__ = 'ENUM'
self.strict = kw.pop('strict', False)
length = max([len(v) for v in self.enums] + [0])
- super(MSEnum, self).__init__(length, **kw)
+ super(ENUM, self).__init__(length=length, **kw)
def bind_processor(self, dialect):
- super_convert = super(MSEnum, self).bind_processor(dialect)
+ super_convert = super(ENUM, self).bind_processor(dialect)
def process(value):
if self.strict and value is not None and value not in self.enums:
raise exc.InvalidRequestError('"%s" not a valid value for '
return value
return process
-class MSSet(MSString):
+class SET(_StringType):
"""MySQL SET type."""
__visit_name__ = 'SET'
self.values = strip_values
length = max([len(v) for v in strip_values] + [0])
- super(MSSet, self).__init__(length, **kw)
+ super(SET, self).__init__(length=length, **kw)
def result_processor(self, dialect):
def process(value):
return process
def bind_processor(self, dialect):
- super_convert = super(MSSet, self).bind_processor(dialect)
+ super_convert = super(SET, self).bind_processor(dialect)
def process(value):
if value is None or isinstance(value, (int, long, basestring)):
pass
return value
return process
-
-class MSBoolean(sqltypes.Boolean):
+class BOOLEAN(sqltypes.Boolean):
"""MySQL BOOLEAN type."""
__visit_name__ = 'BOOLEAN'
return value and True or False
return process
+# old names
+MSBoolean = BOOLEAN
+MSSet = SET
+MSEnum = ENUM
+MSLongBlob = LONGBLOB
+MSMediumBlob = MEDIUMBLOB
+MSTinyBlob = TINYBLOB
+MSBlob = BLOB
+MSBinary = BINARY
+MSVarBinary = VARBINARY
+MSNChar = NCHAR
+MSNVarChar = NVARCHAR
+MSChar = CHAR
+MSString = VARCHAR
+MSLongText = LONGTEXT
+MSMediumText = MEDIUMTEXT
+MSTinyText = TINYTEXT
+MSText = TEXT
+MSYear = YEAR
+MSTimeStamp = TIMESTAMP
+MSTime = TIME
+MSBit = BIT
+MSSmallInteger = SMALLINT
+MSTinyInteger = TINYINT
+MSMediumInteger = MEDIUMINT
+MSBigInteger = BIGINT
+MSNumeric = NUMERIC
+MSDecimal = DECIMAL
+MSDouble = DOUBLE
+MSReal = REAL
+MSFloat = FLOAT
+MSInteger = INTEGER
+
colspecs = {
- sqltypes.Integer: MSInteger,
- sqltypes.SmallInteger: MSSmallInteger,
- sqltypes.Numeric: MSNumeric,
- sqltypes.Float: MSFloat,
- sqltypes.DateTime: MSDateTime,
- sqltypes.Date: MSDate,
- sqltypes.Time: MSTime,
- sqltypes.String: MSString,
- sqltypes.Binary: MSBlob,
- sqltypes.Boolean: MSBoolean,
- sqltypes.Text: MSText,
- sqltypes.CHAR: MSChar,
- sqltypes.NCHAR: MSNChar,
- sqltypes.TIMESTAMP: MSTimeStamp,
- sqltypes.BLOB: MSBlob,
- MSDouble: MSDouble,
- MSReal: MSReal,
- _BinaryType: _BinaryType,
+ sqltypes.Numeric: NUMERIC,
+ sqltypes.Float: FLOAT,
+ sqltypes.Binary: _BinaryType,
+ sqltypes.Boolean: BOOLEAN,
+ sqltypes.Time: TIME,
}
# Everything 3.23 through 5.1 excepting OpenGIS types.
ischema_names = {
- 'bigint': MSBigInteger,
- 'binary': MSBinary,
- 'bit': MSBit,
- 'blob': MSBlob,
- 'boolean':MSBoolean,
- 'char': MSChar,
- 'date': MSDate,
- 'datetime': MSDateTime,
- 'decimal': MSDecimal,
- 'double': MSDouble,
- 'enum': MSEnum,
- 'fixed': MSDecimal,
- 'float': MSFloat,
- 'int': MSInteger,
- 'integer': MSInteger,
- 'longblob': MSLongBlob,
- 'longtext': MSLongText,
- 'mediumblob': MSMediumBlob,
- 'mediumint': MSMediumInteger,
- 'mediumtext': MSMediumText,
- 'nchar': MSNChar,
- 'nvarchar': MSNVarChar,
- 'numeric': MSNumeric,
- 'set': MSSet,
- 'smallint': MSSmallInteger,
- 'text': MSText,
- 'time': MSTime,
- 'timestamp': MSTimeStamp,
- 'tinyblob': MSTinyBlob,
- 'tinyint': MSTinyInteger,
- 'tinytext': MSTinyText,
- 'varbinary': MSVarBinary,
- 'varchar': MSString,
- 'year': MSYear,
+ 'bigint': BIGINT,
+ 'binary': BINARY,
+ 'bit': BIT,
+ 'blob': BLOB,
+ 'boolean':BOOLEAN,
+ 'char': CHAR,
+ 'date': DATE,
+ 'datetime': DATETIME,
+ 'decimal': DECIMAL,
+ 'double': DOUBLE,
+ 'enum': ENUM,
+ 'fixed': DECIMAL,
+ 'float': FLOAT,
+ 'int': INTEGER,
+ 'integer': INTEGER,
+ 'longblob': LONGBLOB,
+ 'longtext': LONGTEXT,
+ 'mediumblob': MEDIUMBLOB,
+ 'mediumint': MEDIUMINT,
+ 'mediumtext': MEDIUMTEXT,
+ 'nchar': NCHAR,
+ 'nvarchar': NVARCHAR,
+ 'numeric': NUMERIC,
+ 'set': SET,
+ 'smallint': SMALLINT,
+ 'text': TEXT,
+ 'time': TIME,
+ 'timestamp': TIMESTAMP,
+ 'tinyblob': TINYBLOB,
+ 'tinyint': TINYINT,
+ 'tinytext': TINYTEXT,
+ 'varbinary': VARBINARY,
+ 'varchar': VARCHAR,
+ 'year': YEAR,
}
class MySQLExecutionContext(default.DefaultExecutionContext):
def visit_typeclause(self, typeclause):
type_ = typeclause.type.dialect_impl(self.dialect)
- if isinstance(type_, MSInteger):
+ if isinstance(type_, sqltypes.Integer):
if getattr(type_, 'unsigned', False):
return 'UNSIGNED INTEGER'
else:
return 'SIGNED INTEGER'
- elif isinstance(type_, (MSDecimal, MSDateTime, MSDate, MSTime)):
+ elif isinstance(type_, sqltypes.TIMESTAMP):
+ return 'DATETIME'
+ elif isinstance(type_, (sqltypes.DECIMAL, sqltypes.DateTime, sqltypes.Date, sqltypes.Time)):
return self.dialect.type_compiler.process(type_)
- elif isinstance(type_, MSText):
+ elif isinstance(type_, sqltypes.Text):
return 'CHAR'
- elif (isinstance(type_, _StringType) and not
- isinstance(type_, (MSEnum, MSSet))):
+ elif (isinstance(type_, sqltypes.String) and not
+ isinstance(type_, (ENUM, SET))):
if getattr(type_, 'length'):
return 'CHAR(%s)' % type_.length
else:
return 'CHAR'
- elif isinstance(type_, _BinaryType):
+ elif isinstance(type_, sqltypes.Binary):
return 'BINARY'
- elif isinstance(type_, MSNumeric):
+ elif isinstance(type_, NUMERIC):
return self.dialect.type_compiler.process(type_).replace('NUMERIC', 'DECIMAL')
- elif isinstance(type_, MSTimeStamp):
- return 'DATETIME'
- elif isinstance(type_, (MSDateTime, MSDate, MSTime)):
- return self.dialect.type_compiler.process(type_)
else:
return None
@reflection.cache
def get_columns(self, connection, table_name, schema=None, **kw):
-
parsed_state = self._parsed_state_or_create(connection, table_name, schema, **kw)
return parsed_state.columns
@reflection.cache
def get_primary_keys(self, connection, table_name, schema=None, **kw):
-
parsed_state = self._parsed_state_or_create(connection, table_name, schema, **kw)
for key in parsed_state.keys:
if key['type'] == 'PRIMARY':
parsed_state = self._parsed_state_or_create(connection, table_name, schema, **kw)
default_schema = None
-
+
fkeys = []
for spec in parsed_state.constraints:
return sql
def _parsed_state_or_create(self, connection, table_name, schema=None, **kw):
- if 'parsed_state' in kw:
- return kw['parsed_state']
- else:
- return self._setup_parser(connection, table_name, schema)
+ return self._setup_parser(
+ connection,
+ table_name,
+ schema,
+ info_cache=kw.get('info_cache', None)
+ )
- def _setup_parser(self, connection, table_name, schema=None):
-
+ @reflection.cache
+ def _setup_parser(self, connection, table_name, schema=None, **kw):
charset = self._connection_charset
try:
parser = self.parser
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
