From 025e4b0e1ba5e2ad4dc15c701154f03b6336dc13 Mon Sep 17 00:00:00 2001 From: Dmitry Maslennikov Date: Fri, 11 Nov 2022 17:41:09 +0400 Subject: [PATCH] historic dates support --- sqlalchemy_iris/base.py | 110 ++------------------------------ sqlalchemy_iris/requirements.py | 15 +++++ sqlalchemy_iris/types.py | 101 +++++++++++++++++++++++++++++ 3 files changed, 121 insertions(+), 105 deletions(-) create mode 100644 sqlalchemy_iris/types.py diff --git a/sqlalchemy_iris/base.py b/sqlalchemy_iris/base.py index 8ce5da9..17f5163 100644 --- a/sqlalchemy_iris/base.py +++ b/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, } diff --git a/sqlalchemy_iris/requirements.py b/sqlalchemy_iris/requirements.py index 0b8ed11..b7a79ed 100644 --- a/sqlalchemy_iris/requirements.py +++ b/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() + diff --git a/sqlalchemy_iris/types.py b/sqlalchemy_iris/types.py new file mode 100644 index 0000000..c18fcf4 --- /dev/null +++ b/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