historic dates support

sqlalchemy_iris/base.py

@@ -1,7 +1,6 @@
-import datetime
-from decimal import Decimal
 import intersystems_iris.dbapi._DBAPI as dbapi
 from . import information_schema as ischema
+from . import types
 from sqlalchemy import exc
 from sqlalchemy.orm import aliased
 from sqlalchemy.engine import default
@@ -539,110 +538,11 @@ def create_cursor(self):
         return cursor
 
 
-HOROLOG_ORDINAL = datetime.date(1840, 12, 31).toordinal()
-
-
-class _IRISDate(sqltypes.Date):
-    def bind_processor(self, dialect):
-        def process(value):
-            if value is None:
-                return None
-            horolog = value.toordinal() - HOROLOG_ORDINAL
-            return str(horolog)
-
-        return process
-
-    def result_processor(self, dialect, coltype):
-        def process(value):
-            if value is None:
-                return None
-            if isinstance(value, str) and '-' in value:
-                return datetime.datetime.strptime(value, '%Y-%m-%d').date()
-            horolog = int(value) + HOROLOG_ORDINAL
-            return datetime.date.fromordinal(horolog)
-
-        return process
-
-
-class _IRISTimeStamp(sqltypes.DateTime):
-    def bind_processor(self, dialect):
-        def process(value: datetime.datetime):
-            if value is not None:
-                # value = int(value.timestamp() * 1000000)
-                # value += (2 ** 60) if value > 0 else -(2 ** 61 * 3)
-                return value.strftime('%Y-%m-%d %H:%M:%S.%f')
-            return value
-
-        return process
-
-    def result_processor(self, dialect, coltype):
-        def process(value):
-            if isinstance(value, str):
-                if '.' not in value:
-                    value += '.0'
-                return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M:%S.%f')
-            if isinstance(value, int):
-                value -= (2 ** 60) if value > 0 else -(2 ** 61 * 3)
-                value = value / 1000000
-                value = datetime.datetime.utcfromtimestamp(value)
-            return value
-
-        return process
-
-
-class _IRISDateTime(sqltypes.DateTime):
-    def bind_processor(self, dialect):
-        def process(value):
-            if value is not None:
-                return value.strftime('%Y-%m-%d %H:%M:%S.%f')
-            return value
-
-        return process
-
-    def result_processor(self, dialect, coltype):
-        def process(value):
-            if isinstance(value, str):
-                if '.' not in value:
-                    value += '.0'
-                return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M:%S.%f')
-            return value
-
-        return process
-
-
-class _IRISTime(sqltypes.DateTime):
-    def bind_processor(self, dialect):
-        def process(value):
-            if value is not None:
-                return value.strftime('%H:%M:%S.%f')
-            return value
-
-        return process
-
-    def result_processor(self, dialect, coltype):
-        def process(value):
-            if isinstance(value, str):
-                if '.' not in value:
-                    value += '.0'
-                return datetime.datetime.strptime(value, '%H:%M:%S.%f').time()
-            if isinstance(value, int) or isinstance(value, Decimal):
-                horolog = value
-                hour = int(horolog // 3600)
-                horolog -= int(hour * 3600)
-                minute = int(horolog // 60)
-                second = int(horolog % 60)
-                micro = int(value % 1 * 1000000)
-                return datetime.time(hour, minute, second, micro)
-            return value
-
-        return process
-
-
 colspecs = {
-    sqltypes.Date: _IRISDate,
-    sqltypes.DateTime: _IRISDateTime,
-    sqltypes.TIMESTAMP: _IRISTimeStamp,
-    sqltypes.Time: _IRISTime,
+    sqltypes.Date: types.IRISDate,
+    sqltypes.DateTime: types.IRISDateTime,
+    sqltypes.TIMESTAMP: types.IRISTimeStamp,
+    sqltypes.Time: types.IRISTime,
 }
 
 

sqlalchemy_iris/requirements.py

@@ -24,3 +24,18 @@ def supports_distinct_on(self):
     @property
     def reflects_pk_names(self):
         return exclusions.open()
+
+    @property
+    def date_historic(self):
+        """target dialect supports representation of Python
+        datetime.datetime() objects with historic (pre 1970) values."""
+
+        return exclusions.open()
+
+    @property
+    def datetime_historic(self):
+        """target dialect supports representation of Python
+        datetime.datetime() objects with historic (pre 1970) values."""
+
+        return exclusions.open()
+

sqlalchemy_iris/types.py

@@ -0,0 +1,101 @@
+import datetime
+from decimal import Decimal
+from sqlalchemy.sql import sqltypes
+
+HOROLOG_ORDINAL = datetime.date(1840, 12, 31).toordinal()
+
+
+class IRISDate(sqltypes.Date):
+    def bind_processor(self, dialect):
+        def process(value):
+            if value is None:
+                return None
+            horolog = value.toordinal() - HOROLOG_ORDINAL
+            return str(horolog)
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if value is None:
+                return None
+            if isinstance(value, str) and '-' in value[1:]:
+                return datetime.datetime.strptime(value, '%Y-%m-%d').date()
+            horolog = int(value) + HOROLOG_ORDINAL
+            return datetime.date.fromordinal(horolog)
+
+        return process
+
+
+class IRISTimeStamp(sqltypes.DateTime):
+    def bind_processor(self, dialect):
+        def process(value: datetime.datetime):
+            if value is not None:
+                # value = int(value.timestamp() * 1000000)
+                # value += (2 ** 60) if value > 0 else -(2 ** 61 * 3)
+                return value.strftime('%Y-%m-%d %H:%M:%S.%f')
+            return value
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if isinstance(value, str):
+                if '.' not in value:
+                    value += '.0'
+                return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M:%S.%f')
+            if isinstance(value, int):
+                value -= (2 ** 60) if value > 0 else -(2 ** 61 * 3)
+                value = value / 1000000
+                value = datetime.datetime.utcfromtimestamp(value)
+            return value
+
+        return process
+
+
+class IRISDateTime(sqltypes.DateTime):
+    def bind_processor(self, dialect):
+        def process(value):
+            if value is not None:
+                return value.strftime('%Y-%m-%d %H:%M:%S.%f')
+            return value
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if isinstance(value, str):
+                if '.' not in value:
+                    value += '.0'
+                return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M:%S.%f')
+            return value
+
+        return process
+
+
+class IRISTime(sqltypes.DateTime):
+    def bind_processor(self, dialect):
+        def process(value):
+            if value is not None:
+                return value.strftime('%H:%M:%S.%f')
+            return value
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if isinstance(value, str):
+                if '.' not in value:
+                    value += '.0'
+                return datetime.datetime.strptime(value, '%H:%M:%S.%f').time()
+            if isinstance(value, int) or isinstance(value, Decimal):
+                horolog = value
+                hour = int(horolog // 3600)
+                horolog -= int(hour * 3600)
+                minute = int(horolog // 60)
+                second = int(horolog % 60)
+                micro = int(value % 1 * 1000000)
+                return datetime.time(hour, minute, second, micro)
+            return value
+
+        return process