from sqlalchemy import processors, types as sqltypes
class PyODBCNumeric(sqltypes.Numeric):
- """Turns Decimals with adjusted() < -6 into floats."""
+ """Turns Decimals with adjusted() < -6 into floats, > 7 into strings"""
def bind_processor(self, dialect):
super_process = super(PyODBCNumeric, self).bind_processor(dialect)
def process(value):
if self.asdecimal and \
- isinstance(value, decimal.Decimal) and \
- value.adjusted() < -6:
- return processors.to_float(value)
- elif super_process:
+ isinstance(value, decimal.Decimal):
+
+ if value.adjusted() < -6:
+ return processors.to_float(value)
+ elif value.adjusted() > 7:
+ return self._large_dec_to_string(value)
+
+ if super_process:
return super_process(value)
else:
return value
return process
- # This method turns the adjusted into a string.
- # not sure if this has advantages over the simple float
- # approach above.
-# def bind_processor(self, dialect):
-# def process(value):
-# if isinstance(value, decimal.Decimal):
-# if value.adjusted() < 0:
-# result = "%s0.%s%s" % (
-# (value < 0 and '-' or ''),
-# '0' * (abs(value.adjusted()) - 1),
-# "".join([str(nint) for nint in value._int]))
-#
-# else:
-# if 'E' in str(value):
-# result = "%s%s%s" % (
-# (value < 0 and '-' or ''),
-# "".join([str(s) for s in value._int]),
-# "0" * (value.adjusted() - (len(value._int)-1)))
-# else:
-# if (len(value._int) - 1) > value.adjusted():
-# result = "%s%s.%s" % (
-# (value < 0 and '-' or ''),
-# "".join([str(s) for s in value._int][0:value.adjusted() + 1]),
-# "".join([str(s) for s in value._int][value.adjusted() + 1:]))
-# else:
-# result = "%s%s" % (
-# (value < 0 and '-' or ''),
-# "".join([str(s) for s in value._int][0:value.adjusted() + 1]))
-# return result
-#
-# else:
-# return value
-# return process
+ def _large_dec_to_string(self, value):
+ if 'E' in str(value):
+ result = "%s%s%s" % (
+ (value < 0 and '-' or ''),
+ "".join([str(s) for s in value._int]),
+ "0" * (value.adjusted() - (len(value._int)-1)))
+ else:
+ if (len(value._int) - 1) > value.adjusted():
+ result = "%s%s.%s" % (
+ (value < 0 and '-' or ''),
+ "".join([str(s) for s in value._int][0:value.adjusted() + 1]),
+ "".join([str(s) for s in value._int][value.adjusted() + 1:]))
+ else:
+ result = "%s%s" % (
+ (value < 0 and '-' or ''),
+ "".join([str(s) for s in value._int][0:value.adjusted() + 1]))
+ return result
class PyODBCConnector(Connector):
driver='pyodbc'
columns = [
# column type, args, kwargs, expected ddl
- (mssql.MSNumeric, [], {},
+ (types.NUMERIC, [], {},
'NUMERIC'),
- (mssql.MSNumeric, [None], {},
+ (types.NUMERIC, [None], {},
'NUMERIC'),
- (mssql.MSNumeric, [12, 4], {},
+ (types.NUMERIC, [12, 4], {},
'NUMERIC(12, 4)'),
(types.Float, [], {},
numbers,
numbers
)
+
+ @testing.fails_on("sybase+pyodbc",
+ "Don't know how do get these values through FreeTDS + Sybase")
+ @testing.fails_on('sqlite', 'The 319438... number fails.')
+ def test_enotation_decimal_large(self):
+ """test exceedingly large decimals.
+
+ """
+
+ numbers = set([
+ decimal.Decimal('4E+8'),
+ decimal.Decimal("31943874831932418390.01"),
+ decimal.Decimal("5748E+15"),
+ decimal.Decimal('1.521E+15'),
+ decimal.Decimal('00000000000000.1E+12'),
+ ])
+ self._do_test(
+ Numeric(precision=25, scale=2),
+ numbers,
+ numbers
+ )