Support Interval Type

Author: rad-pat

Pipfile.lock

@@ -48,6 +48,8 @@ def escape_item(self, item):
  return self.escape_number(item)
  elif isinstance(item, decimal.Decimal):
  return self.escape_number(item)
+ elif isinstance(item, timedelta):
+ return self.escape_string(f'{item.total_seconds()} seconds') + '::interval'
  elif isinstance(item, (datetime, date, time, timedelta)):
  return self.escape_string(item.strftime("%Y-%m-%d %H:%M:%S"))
  else:

databend_sqlalchemy/connector.py

@@ -50,7 +50,9 @@
 )
 from sqlalchemy.engine import ExecutionContext, default
 from sqlalchemy.exc import DBAPIError, NoSuchTableError
+
 from .dml import Merge
+from .types import INTERVAL
 
 RESERVED_WORDS = {
  'Error', 'EOI', 'Whitespace', 'Comment', 'CommentBlock', 'Ident', 'ColumnPosition', 'LiteralString',
@@ -232,44 +234,8 @@ def process(value):
  return process
 
 
-class DatabendInterval(sqltypes.Interval):
- """Stores interval as a datetime relative to epoch, see base implementation."""
-
- _reg = re.compile(r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)")
-
- def result_processor(self, dialect, coltype):
- def process(value):
- if value is None:
- return None
- if isinstance(value, str):
- m = self._reg.match(value)
- if not m:
- raise ValueError(
- "could not parse %r as a datetime value" % (value,)
- )
- groups = m.groups()
- dt = datetime.datetime(*[
- int(groups[0] or self.epoch.year),
- int(groups[1] or self.epoch.month),
- int(groups[2] or self.epoch.day),
- int(groups[3] or 0),
- int(groups[4] or 0),
- int(groups[5] or 0),
- ])
- else:
- dt = value
- return dt - self.epoch
-
- return process
-
- def literal_processor(self, dialect):
- def process(value):
- if value is not None:
- d = self.epoch + value
- interval_str = d.isoformat(" ", timespec="microseconds")
- return f"'{interval_str}'"
-
- return process
+class DatabendInterval(INTERVAL):
+ render_bind_cast = True
 
 
 # Type converters
@@ -526,6 +492,9 @@ def visit_JSON(self, type_, **kw):
  def visit_TIME(self, type_, **kw):
  return "DATETIME"
 
+ def visit_INTERVAL(self, type, **kw):
+ return "INTERVAL"
+
 
 class DatabendDDLCompiler(compiler.DDLCompiler):
 

databend_sqlalchemy/databend_dialect.py

@@ -1,7 +1,7 @@
 
 from sqlalchemy.testing.provision import create_db
 from sqlalchemy.testing.provision import drop_db
-from sqlalchemy.testing.provision import configure_follower
+from sqlalchemy.testing.provision import configure_follower, update_db_opts
 
 
 @create_db.for_db("databend")
@@ -36,3 +36,8 @@ def _databend_drop_db(cfg, eng, ident):
 def _databend_configure_follower(config, ident):
  config.test_schema = "%s_test_schema" % ident
  config.test_schema_2 = "%s_test_schema_2" % ident
+
+# Uncomment to debug SQL Statements in tests
+# @update_db_opts.for_db("databend")
+# def _mssql_update_db_opts(db_url, db_opts):
+# db_opts["echo"] = True

databend_sqlalchemy/provision.py

@@ -0,0 +1,74 @@
+
+from __future__ import annotations
+
+import datetime as dt
+from typing import Optional, Type, Any
+
+from sqlalchemy.engine.interfaces import Dialect
+from sqlalchemy.sql import sqltypes
+from sqlalchemy.sql import type_api
+
+
+# ToDo - This is perhaps how numeric should be defined
+# class NUMERIC(sqltypes.Numeric):
+# def result_processor(self, dialect, type_):
+#
+# orig = super().result_processor(dialect, type_)
+#
+# def process(value):
+# if value is not None:
+# if self.decimal_return_scale:
+# value = decimal.Decimal(f'{value:.{self.decimal_return_scale}f}')
+# else:
+# value = decimal.Decimal(value)
+# if orig:
+# return orig(value)
+# return value
+#
+# return process
+
+
+class INTERVAL(type_api.NativeForEmulated, sqltypes._AbstractInterval):
+ """Databend INTERVAL type."""
+
+ __visit_name__ = "INTERVAL"
+ native = True
+
+ def __init__(
+ self, precision: Optional[int] = None, fields: Optional[str] = None
+ ) -> None:
+ """Construct an INTERVAL.
+
+ :param precision: optional integer precision value
+ :param fields: string fields specifier. allows storage of fields
+ to be limited, such as ``"YEAR"``, ``"MONTH"``, ``"DAY TO HOUR"``,
+ etc.
+
+ """
+ self.precision = precision
+ self.fields = fields
+
+ @classmethod
+ def adapt_emulated_to_native(
+ cls, interval: sqltypes.Interval, **kw: Any # type: ignore[override]
+ ) -> INTERVAL:
+ return INTERVAL(precision=interval.second_precision)
+
+ @property
+ def _type_affinity(self) -> Type[sqltypes.Interval]:
+ return sqltypes.Interval
+
+ def as_generic(self, allow_nulltype: bool = False) -> sqltypes.Interval:
+ return sqltypes.Interval(native=True, second_precision=self.precision)
+
+ @property
+ def python_type(self) -> Type[dt.timedelta]:
+ return dt.timedelta
+
+ def literal_processor(
+ self, dialect: Dialect
+ ) -> Optional[type_api._LiteralProcessorType[dt.timedelta]]:
+ def process(value: dt.timedelta) -> str:
+ return f"to_interval('{value.total_seconds()} seconds')"
+
+ return process

databend_sqlalchemy/types.py

@@ -27,7 +27,7 @@ project_urls =
 [options]
 packages = find:
 install_requires =
- databend-driver>=0.16.0
+ databend-driver>=0.25.0
  sqlalchemy>=1.4
 python_requires = >=3.7
 package_dir =

setup.cfg

@@ -16,13 +16,10 @@
 from sqlalchemy.testing.suite import JoinTest as _JoinTest
 from sqlalchemy.testing.suite import BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest
 from sqlalchemy.testing.suite import ServerSideCursorsTest as _ServerSideCursorsTest
-from sqlalchemy.testing.suite import IntervalTest as _IntervalTest
-from sqlalchemy.testing.suite import PrecisionIntervalTest as _PrecisionIntervalTest
 from sqlalchemy.testing.suite import EnumTest as _EnumTest
 from sqlalchemy import types as sql_types
 from sqlalchemy import testing, select
 from sqlalchemy.testing import config, eq_
-from databend_sqlalchemy.databend_dialect import DatabendInterval
 
 
 class ComponentReflectionTest(_ComponentReflectionTest):
@@ -283,47 +280,6 @@ def test_roundtrip_fetchmany(self):
  pass
 
 
-class IntervalTest(_IntervalTest):
- __backend__ = True
- datatype = DatabendInterval
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_literal_interval(self, connection):
- pass
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_table_interval_and_literal_interval(
- self, connection, arithmetic_table_fixture
- ):
- pass
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_table_date_and_literal_interval(
- self, connection, arithmetic_table_fixture
- ):
- pass
-
-
-class PrecisionIntervalTest(_PrecisionIntervalTest):
- __backend__ = True
- datatype = DatabendInterval
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_literal_interval(self, connection):
- pass
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_table_interval_and_literal_interval(
- self, connection, arithmetic_table_fixture
- ):
- pass
-
- @testing.skip("databend") # Skipped because cannot figure out the literal() part
- def test_arithmetic_operation_table_date_and_literal_interval(
- self, connection, arithmetic_table_fixture
- ):
- pass
-
 class EnumTest(_EnumTest):
  __backend__ = True