supports_sane_multi_rowcount = False
supports_unicode_statements = False
supports_unicode_binds = False
- supports_native_decimal = False
+
+ supports_native_decimal = True
@classmethod
def dbapi(cls):
"""
-import datetime, decimal, re
+import datetime, re
from sqlalchemy import schema as sa_schema
from sqlalchemy import exc, types as sqltypes, sql, util
class _MSNumeric(sqltypes.Numeric):
- def result_processor(self, dialect, coltype):
- if self.asdecimal:
- # TODO: factor this down into the sqltypes.Numeric class,
- # use dialect flags
- if getattr(self, 'scale', None) is None:
- # we're a "float". return a default decimal factory
- return processors.to_decimal_processor_factory(decimal.Decimal)
- elif dialect.supports_native_decimal:
- # we're a "numeric", DBAPI will give us Decimal directly
- return None
- else:
- # we're a "numeric", DBAPI returns floats, convert.
- return processors.to_decimal_processor_factory(decimal.Decimal, self.scale)
- else:
- #XXX: if the DBAPI returns a float (this is likely, given the
- # processor when asdecimal is True), this should be a None
- # processor instead.
- return processors.to_float
-
+
def bind_processor(self, dialect):
def process(value):
+ # TODO: this seems exceedingly complex.
+ # need to know exactly what tests cover this, so far
+ # test_types.NumericTest.test_enotation_decimal
+
if isinstance(value, decimal.Decimal):
if value.adjusted() < 0:
result = "%s0.%s%s" % (
from sqlalchemy.dialects.mysql.base import (MySQLDialect,
MySQLExecutionContext, MySQLCompiler, MySQLIdentifierPreparer,
- BIT, NUMERIC, _NumericType)
+ BIT)
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy.sql import operators as sql_operators
def post_process_text(self, text):
return text.replace('%', '%%')
-class _DecimalType(_NumericType):
- def result_processor(self, dialect, coltype):
- if self.asdecimal:
- return None
- return processors.to_float
-
-class _myconnpyNumeric(_DecimalType, NUMERIC):
- pass
-
class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
def _escape_identifier(self, value):
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
+ supports_native_decimal = True
+
default_paramstyle = 'format'
execution_ctx_cls = MySQLExecutionContext_mysqlconnector
statement_compiler = MySQLCompiler_mysqlconnector
colspecs = util.update_copy(
MySQLDialect.colspecs,
{
- sqltypes.Numeric: _myconnpyNumeric,
BIT: _myconnpyBIT,
}
)
create_engine('mysql:///mydb?charset=utf8&use_unicode=0')
"""
-import decimal
import re
-from sqlalchemy.dialects.mysql.base import (DECIMAL, MySQLDialect, MySQLExecutionContext,
- MySQLCompiler, MySQLIdentifierPreparer, NUMERIC, _NumericType)
+from sqlalchemy.dialects.mysql.base import (MySQLDialect, MySQLExecutionContext,
+ MySQLCompiler, MySQLIdentifierPreparer)
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy import exc, log, schema, sql, types as sqltypes, util
return text.replace('%', '%%')
-class _DecimalType(_NumericType):
- def result_processor(self, dialect, coltype):
- if self.asdecimal:
- return None
- return processors.to_float
-
-
-class _MySQLdbNumeric(_DecimalType, NUMERIC):
- pass
-
-
-class _MySQLdbDecimal(_DecimalType, DECIMAL):
- pass
-
class MySQLIdentifierPreparer_mysqldb(MySQLIdentifierPreparer):
def _escape_identifier(self, value):
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
+ supports_native_decimal = True
+
default_paramstyle = 'format'
execution_ctx_cls = MySQLExecutionContext_mysqldb
statement_compiler = MySQLCompiler_mysqldb
colspecs = util.update_copy(
MySQLDialect.colspecs,
{
- sqltypes.Numeric: _MySQLdbNumeric,
- DECIMAL: _MySQLdbDecimal
}
)
create_engine('mysql+oursql:///mydb?charset=latin1')
"""
-import decimal
import re
from sqlalchemy.dialects.mysql.base import (BIT, MySQLDialect, MySQLExecutionContext,
- MySQLCompiler, MySQLIdentifierPreparer, NUMERIC, _NumericType)
+ MySQLCompiler, MySQLIdentifierPreparer)
from sqlalchemy.engine import base as engine_base, default
from sqlalchemy.sql import operators as sql_operators
from sqlalchemy import exc, log, schema, sql, types as sqltypes, util
from sqlalchemy import processors
-class _oursqlNumeric(NUMERIC):
- def result_processor(self, dialect, coltype):
- if self.asdecimal:
- return None
- return processors.to_float
-
class _oursqlBIT(BIT):
def result_processor(self, dialect, coltype):
supports_unicode_binds = True
supports_unicode_statements = True
# end Py2K
-
+
+ supports_native_decimal = True
+
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
execution_ctx_cls = MySQLExecutionContext_oursql
MySQLDialect.colspecs,
{
sqltypes.Time: sqltypes.Time,
- sqltypes.Numeric: _oursqlNumeric,
BIT: _oursqlBIT,
}
)
TIMESTAMP, VARCHAR
-class _NumericMixin(object):
- def bind_processor(self, dialect):
- if self.asdecimal:
- return processors.to_str
- else:
- return processors.to_float
-
-class _SLNumeric(_NumericMixin, sqltypes.Numeric):
- pass
-
-class _SLFloat(_NumericMixin, sqltypes.Float):
- pass
-
class _DateTimeMixin(object):
_reg = None
_storage_format = None
colspecs = {
sqltypes.Date: DATE,
sqltypes.DateTime: DATETIME,
- sqltypes.Float: _SLFloat,
- sqltypes.Numeric: _SLNumeric,
sqltypes.Time: TIME,
}
return dbapi.NUMBER
def bind_processor(self, dialect):
- return processors.to_float
+ if dialect.supports_native_decimal:
+ return None
+ else:
+ return processors.to_float
def result_processor(self, dialect, coltype):
if self.asdecimal:
- #XXX: use decimal from http://www.bytereef.org/libmpdec.html
-# try:
-# from fastdec import mpd as Decimal
-# except ImportError:
- if self.scale is not None:
- return processors.to_decimal_processor_factory(_python_Decimal, self.scale)
+ if dialect.supports_native_decimal:
+ # we're a "numeric", DBAPI will give us Decimal directly
+ return None
else:
- return processors.to_decimal_processor_factory(_python_Decimal)
+ # we're a "numeric", DBAPI returns floats, convert.
+ return processors.to_decimal_processor_factory(_python_Decimal, self.scale)
else:
- return None
+ if dialect.supports_native_decimal:
+ return processors.to_float
+ else:
+ return None
@util.memoized_property
def _expression_adaptations(self):
def result_processor(self, dialect, coltype):
if self.asdecimal:
- #XXX: use decimal from http://www.bytereef.org/libmpdec.html
-# try:
-# from fastdec import mpd as Decimal
-# except ImportError:
return processors.to_decimal_processor_factory(_python_Decimal)
else:
return None
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.test import *
from sqlalchemy.test.util import picklers
+from decimal import Decimal
+from sqlalchemy.test.util import round_decimal
class AdaptTest(TestBase):
foo.create()
foo.drop()
-def _missing_decimal():
- """Python implementation supports decimals"""
- try:
- import decimal
- return False
- except ImportError:
- return True
-
class NumericTest(TestBase, AssertsExecutionResults):
@classmethod
def setup_class(cls):
def teardown(self):
numeric_table.delete().execute()
- @testing.fails_if(_missing_decimal)
def test_decimal(self):
from decimal import Decimal
numeric_table.insert().execute(
(2, 3.5, 5.6, Decimal("12.4"), Decimal("15.75")),
])
- @testing.fails_if(_missing_decimal)
def test_precision_decimal(self):
- from decimal import Decimal
- from sqlalchemy.test.util import round_decimal
t = Table('t', MetaData(), Column('x', Numeric(precision=18, scale=12)))
t.create(testing.db)
ret = set([row[0] for row in testing.db.execute(t.select()).fetchall()])
- numbers = set(round_decimal(n, 11) for n in numbers)
- ret = set(round_decimal(n, 11) for n in ret)
+ if testing.against('sqlite'):
+ numbers = set(round_decimal(n, 11) for n in numbers)
+ ret = set(round_decimal(n, 11) for n in ret)
+ else:
+ numbers = set(n for n in numbers)
+ ret = set(n for n in ret)
eq_(numbers, ret)
finally:
t.drop(testing.db)
+
+ @testing.fails_on('sybase', "Driver doesn't appear to handle E notation, won't accept strings")
+ def test_enotation_decimal(self):
+ """test exceedingly small decimals.
+
+ Decimal reports values with E notation when the exponent
+ is greater than 6.
+
+ """
+
+ t = Table('t', MetaData(), Column('x', Numeric(precision=18, scale=12)))
+ t.create(testing.db)
+ try:
+ numbers = set([
+ decimal.Decimal('1E-2'),
+ decimal.Decimal('1E-3'),
+ decimal.Decimal('1E-4'),
+ decimal.Decimal('1E-5'),
+ decimal.Decimal('1E-6'),
+ decimal.Decimal('1E-7'),
+ decimal.Decimal('1E-8'),
+ ])
+
+ testing.db.execute(t.insert(), [{'x':x} for x in numbers])
+
+ ret = set([row[0] for row in testing.db.execute(t.select()).fetchall()])
+ numbers = set(n for n in numbers)
+ ret = set(n for n in ret)
+
+ eq_(numbers, ret)
+ finally:
+ t.drop(testing.db)
+
def test_decimal_fallback(self):
from decimal import Decimal
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
