- an NCLOB type is added to the base types.
+ - added native INTERVAL type to the dialect. This supports
+ only the DAY TO SECOND interval type so far due to lack
+ of support in cx_oracle for YEAR TO MONTH. [ticket:1467]
+
- usage of the CHAR type results in cx_oracle's
FIXED_CHAR dbapi type being bound to statements.
constraint to enforce the enum.
[ticket:1109] [ticket:1511]
+ - The Interval type includes a "native" flag which controls
+ if native INTERVAL types (postgresql + oracle) are selected
+ if available, or not. "day_precision" and "second_precision"
+ arguments are also added which propagate as appropriately
+ to these native types. Related to [ticket:1467].
+
- The Boolean type, when used on a backend that doesn't
have native boolean support, will generate a CHECK
constraint "col IN (0, 1)" along with the int/smallint-
from sqlalchemy.dialects.oracle.base import \
VARCHAR, NVARCHAR, CHAR, DATE, DATETIME, NUMBER,\
BLOB, BFILE, CLOB, NCLOB, TIMESTAMP, RAW,\
- FLOAT, DOUBLE_PRECISION, LONG, dialect
+ FLOAT, DOUBLE_PRECISION, LONG, dialect, INTERVAL
__all__ = (
'VARCHAR', 'NVARCHAR', 'CHAR', 'DATE', 'DATETIME', 'NUMBER',
'BLOB', 'BFILE', 'CLOB', 'NCLOB', 'TIMESTAMP', 'RAW',
-'FLOAT', 'DOUBLE_PRECISION', 'LONG', 'dialect'
+'FLOAT', 'DOUBLE_PRECISION', 'LONG', 'dialect', 'INTERVAL'
)
class LONG(sqltypes.Text):
__visit_name__ = 'LONG'
+
+class INTERVAL(sqltypes.TypeEngine):
+ __visit_name__ = 'INTERVAL'
+
+ def __init__(self,
+ day_precision=None,
+ second_precision=None):
+ """Construct an INTERVAL.
+
+ Note that only DAY TO SECOND intervals are currently supported.
+ This is due to a lack of support for YEAR TO MONTH intervals
+ within available DBAPIs (cx_oracle and zxjdbc).
+
+ :param day_precision: the day precision value. this is the number of digits
+ to store for the day field. Defaults to "2"
+ :param second_precision: the second precision value. this is the number of digits
+ to store for the fractional seconds field. Defaults to "6".
+
+ """
+ self.day_precision = day_precision
+ self.second_precision = second_precision
+
+ @classmethod
+ def _adapt_from_generic_interval(cls, interval):
+ return INTERVAL(day_precision=interval.day_precision,
+ second_precision=interval.second_precision)
+
+ def adapt(self, impltype):
+ return impltype(day_precision=self.day_precision,
+ second_precision=self.second_precision)
+
+ @property
+ def _type_affinity(self):
+ return sqltypes.Interval
class _OracleBoolean(sqltypes.Boolean):
def get_dbapi_type(self, dbapi):
colspecs = {
sqltypes.Boolean : _OracleBoolean,
+ sqltypes.Interval : INTERVAL,
}
ischema_names = {
def visit_unicode(self, type_):
return self.visit_NVARCHAR(type_)
-
+
+ def visit_INTERVAL(self, type_):
+ return "INTERVAL DAY%s TO SECOND%s" % (
+ type_.day_precision is not None and
+ "(%d)" % type_.day_precision or
+ "",
+ type_.second_precision is not None and
+ "(%d)" % type_.second_precision or
+ "",
+ )
+
def visit_DOUBLE_PRECISION(self, type_):
return self._generate_numeric(type_, "DOUBLE PRECISION")
self.implicit_returning = self.server_version_info > (10, ) and \
self.__dict__.get('implicit_returning', True)
+ if self.server_version_info < (9,):
+ self.colspecs = self.colspecs.copy()
+ self.colspecs.pop(sqltypes.Interval)
+
def do_release_savepoint(self, connection, name):
# Oracle does not support RELEASE SAVEPOINT
pass
def bind_processor(self, dialect):
return None
-
+class _OracleInterval(oracle.INTERVAL):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.INTERVAL
+
class _OracleRaw(oracle.RAW):
pass
-
colspecs = {
sqltypes.Date : _OracleDate,
sqltypes.Binary : _OracleBinary,
sqltypes.Boolean : oracle._OracleBoolean,
+ sqltypes.Interval : _OracleInterval,
+ oracle.INTERVAL : _OracleInterval,
sqltypes.Text : _OracleText,
sqltypes.UnicodeText : _OracleUnicodeText,
sqltypes.CHAR : _OracleChar,
def adapt(self, impltype):
return impltype(self.precision)
+ @classmethod
+ def _adapt_from_generic_interval(cls, interval):
+ return INTERVAL(precision=interval.second_precision)
+
@property
def _type_affinity(self):
return sqltypes.Interval
supports_default_values = False
supports_empty_insert = True
+ server_version_info = None
+
# indicates symbol names are
# UPPERCASEd if they are case insensitive
# within the database.
cursor.close()
return result
- @classmethod
- def type_descriptor(cls, typeobj):
+ def type_descriptor(self, typeobj):
"""Provide a database-specific ``TypeEngine`` object, given
the generic object which comes from the types module.
and passes on to ``types.adapt_type()``.
"""
- return sqltypes.adapt_type(typeobj, cls.colspecs)
+ return sqltypes.adapt_type(typeobj, self.colspecs)
def reflecttable(self, connection, table, include_columns):
insp = reflection.Inspector.from_engine(connection)
schema.types = expression.sqltypes =sys.modules['sqlalchemy.types']
from sqlalchemy.util import pickle
from sqlalchemy.sql.visitors import Visitable
-import sqlalchemy.util as util
+from sqlalchemy import util
NoneType = type(None)
if util.jython:
import array
return {}
def dialect_impl(self, dialect, **kwargs):
+ key = (dialect.__class__, dialect.server_version_info)
+
try:
- return self._impl_dict[dialect.__class__]
+ return self._impl_dict[key]
except KeyError:
- return self._impl_dict.setdefault(dialect.__class__, dialect.__class__.type_descriptor(self))
+ return self._impl_dict.setdefault(key, dialect.type_descriptor(self))
def __getstate__(self):
d = self.__dict__.copy()
return cls()
def dialect_impl(self, dialect):
+ key = (dialect.__class__, dialect.server_version_info)
try:
- return self._impl_dict[dialect.__class__]
- except AttributeError:
- self._impl_dict = {}
+ return self._impl_dict[key]
except KeyError:
pass
# adapt the TypeDecorator first, in
# the case that the dialect maps the TD
# to one of its native types (i.e. PGInterval)
- adapted = dialect.__class__.type_descriptor(self)
+ adapted = dialect.type_descriptor(self)
if adapted is not self:
- self._impl_dict[dialect] = adapted
+ self._impl_dict[key] = adapted
return adapted
# otherwise adapt the impl type, link
raise AssertionError("Type object %s does not properly implement the copy() "
"method, it must return an object of type %s" % (self, self.__class__))
tt.impl = typedesc
- self._impl_dict[dialect] = tt
+ self._impl_dict[key] = tt
return tt
@util.memoized_property
if isinstance(self.impl, TypeDecorator):
return self.impl.dialect_impl(dialect)
else:
- return dialect.__class__.type_descriptor(self.impl)
+ return dialect.type_descriptor(self.impl)
def __getattr__(self, key):
"""Proxy all other undefined accessors to the underlying implementation."""
def copy(self):
instance = self.__class__.__new__(self.__class__)
instance.__dict__.update(self.__dict__)
+ instance._impl_dict = {}
return instance
def get_dbapi_type(self, dbapi):
self._on_metadata_create)
table.metadata.append_ddl_listener('after-drop',
self._on_metadata_drop)
-
@property
def bind(self):
impl = DateTime
epoch = dt.datetime.utcfromtimestamp(0)
+ def __init__(self, native=True,
+ second_precision=None,
+ day_precision=None):
+ """Construct an Interval object.
+
+ :param native: when True, use the actual
+ INTERVAL type provided by the database, if
+ supported (currently Postgresql, Oracle).
+ Otherwise, represent the interval data as
+ an epoch value regardless.
+
+ :param second_precision: For native interval types
+ which support a "fractional seconds precision" parameter,
+ i.e. Oracle and Postgresql
+
+ :param day_precision: for native interval types which
+ support a "day precision" parameter, i.e. Oracle.
+
+ """
+ super(Interval, self).__init__()
+ self.native = native
+ self.second_precision = second_precision
+ self.day_precision = day_precision
+
+ def adapt(self, cls):
+ if self.native:
+ return cls._adapt_from_generic_interval(self)
+ else:
+ return self
+
def bind_processor(self, dialect):
impl_processor = self.impl.bind_processor(dialect)
epoch = self.epoch
if has_kw:
stack.update(class_.__bases__)
args.discard('self')
- return list(args)
+ return args
def get_func_kwargs(func):
"""Return the full set of legal kwargs for the given `func`."""
from sqlalchemy.engine import default
from sqlalchemy.util import jython
from decimal import Decimal
+import datetime
import os
class TypesTest(TestBase, AssertsCompiledSQL):
__only_on__ = 'oracle'
+ __dialect__ = oracle.OracleDialect()
def test_no_clobs_for_string_params(self):
"""test that simple string params get a DBAPI type of VARCHAR, not CLOB.
finally:
t1.drop()
+ def test_interval(self):
+
+ for type_, expected in [
+ (oracle.INTERVAL(), "INTERVAL DAY TO SECOND"),
+ (oracle.INTERVAL(day_precision=3), "INTERVAL DAY(3) TO SECOND"),
+ (oracle.INTERVAL(second_precision=5), "INTERVAL DAY TO SECOND(5)"),
+ (oracle.INTERVAL(day_precision=2, second_precision=5), "INTERVAL DAY(2) TO SECOND(5)"),
+ ]:
+ self.assert_compile(type_, expected)
+
+ metadata = MetaData(testing.db)
+ interval_table = Table("intervaltable", metadata,
+ Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
+ Column("day_interval", oracle.INTERVAL(day_precision=3)),
+ )
+ metadata.create_all()
+ try:
+ interval_table.insert().execute(
+ day_interval=datetime.timedelta(days=35, seconds=5743),
+ )
+ row = interval_table.select().execute().first()
+ eq_(row['day_interval'], datetime.timedelta(days=35, seconds=5743))
+ finally:
+ metadata.drop_all()
+
def test_numerics(self):
m = MetaData(testing.db)
t1 = Table('t1', m,
from sqlalchemy.test.testing import eq_
import sqlalchemy.engine.url as url
from sqlalchemy.databases import *
-
+from sqlalchemy.test.schema import Table, Column
from sqlalchemy.test import *
global interval_table, metadata
metadata = MetaData(testing.db)
interval_table = Table("intervaltable", metadata,
- Column("id", Integer, Sequence('interval_id_seq', optional=True), primary_key=True),
- Column("interval", Interval),
+ Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
+ Column("native_interval", Interval()),
+ Column("native_interval_args", Interval(day_precision=3, second_precision=6)),
+ Column("non_native_interval", Interval(native=False)),
)
metadata.create_all()
@testing.fails_on("+pg8000", "Not yet known how to pass values of the INTERVAL type")
@testing.fails_on("postgresql+zxjdbc", "Not yet known how to pass values of the INTERVAL type")
def test_roundtrip(self):
- delta = datetime.datetime(2006, 10, 5) - datetime.datetime(2005, 8, 17)
- interval_table.insert().execute(interval=delta)
- assert interval_table.select().execute().first()['interval'] == delta
+ small_delta = datetime.timedelta(days=15, seconds=5874)
+ delta = datetime.timedelta(414)
+ interval_table.insert().execute(
+ native_interval=small_delta,
+ native_interval_args=delta,
+ non_native_interval=delta
+ )
+ row = interval_table.select().execute().first()
+ eq_(row['native_interval'], small_delta)
+ eq_(row['native_interval_args'], delta)
+ eq_(row['non_native_interval'], delta)
def test_null(self):
- interval_table.insert().execute(id=1, inverval=None)
- assert interval_table.select().execute().first()['interval'] is None
+ interval_table.insert().execute(id=1, native_inverval=None, non_native_interval=None)
+ row = interval_table.select().execute().first()
+ eq_(row['native_interval'], None)
+ eq_(row['native_interval_args'], None)
+ eq_(row['non_native_interval'], None)
class BooleanTest(TestBase, AssertsExecutionResults):
@classmethod
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
