UUID as %UniqeIdentifer in IRIS

Author: daimor

sqlalchemy_iris/base.py

@@ -90,6 +90,7 @@ def check_constraints(cls):
 from .types import IRISTimeStamp
 from .types import IRISDate
 from .types import IRISDateTime
+from .types import IRISUniqueIdentifier
 
 
 ischema_names = {
@@ -748,6 +749,9 @@ def visit_DOUBLE(self, type_, **kw):
     def visit_TINYINT(self, type_, **kw):
         return "TINYINT"
 
+    def visit_UUID(self, type_, **kw):
+        return "%UniqueIdentifier"
+
 
 class IRISIdentifierPreparer(sql.compiler.IdentifierPreparer):
     """Install IRIS specific reserved words."""
@@ -804,6 +808,7 @@ def create_cursor(self):
     sqltypes.DateTime: IRISDateTime,
     sqltypes.TIMESTAMP: IRISTimeStamp,
     sqltypes.Time: IRISTime,
+    sqltypes.UUID: IRISUniqueIdentifier,
 }
 
 

sqlalchemy_iris/requirements.py

@@ -19,7 +19,7 @@ def array_type(self):
 
     @property
     def uuid_data_type(self):
-        return exclusions.closed()
+        return exclusions.open()
 
     @property
     def check_constraints(self):

sqlalchemy_iris/types.py

@@ -1,6 +1,7 @@
 import datetime
 from decimal import Decimal
 from sqlalchemy.sql import sqltypes
+from uuid import UUID as _python_UUID
 
 HOROLOG_ORDINAL = datetime.date(1840, 12, 31).toordinal()
 
@@ -16,13 +17,15 @@ def process(value):
             elif isinstance(value, bool):
                 return 1 if value is True else 0
             return None
+
         return process
 
     def result_processor(self, dialect, coltype):
         def process(value):
             if isinstance(value, int):
                 return value > 0
             return value
+
         return process
 
 
@@ -40,8 +43,8 @@ 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()
+            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)
 
@@ -54,19 +57,19 @@ 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.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 "." 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 -= (2**60) if value > 0 else -(2**61 * 3)
                 value = value / 1000000
                 value = datetime.datetime.utcfromtimestamp(value)
             return value
@@ -78,17 +81,17 @@ 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.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 "." not in value:
+                    value += ".0"
+                return datetime.datetime.strptime(value, "%Y-%m-%d %H:%M:%S.%f")
             return value
 
         return process
@@ -98,17 +101,17 @@ 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.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 "." 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)
@@ -122,6 +125,75 @@ def process(value):
         return process
 
 
+class IRISUniqueIdentifier(sqltypes.Uuid):
+    def literal_processor(self, dialect):
+        if not self.as_uuid:
+
+            def process(value):
+                return f"""'{value.replace("'", "''")}'"""
+
+            return process
+        else:
+
+            def process(value):
+                return f"""'{str(value).replace("'", "''")}'"""
+
+            return process
+
+    def bind_processor(self, dialect):
+        character_based_uuid = not dialect.supports_native_uuid or not self.native_uuid
+
+        if character_based_uuid:
+            if self.as_uuid:
+
+                def process(value):
+                    if value is not None:
+                        value = str(value)
+                    return value
+
+                return process
+            else:
+
+                def process(value):
+                    return value
+
+                return process
+        else:
+            return None
+
+    def result_processor(self, dialect, coltype):
+        character_based_uuid = not dialect.supports_native_uuid or not self.native_uuid
+
+        if character_based_uuid:
+            if self.as_uuid:
+
+                def process(value):
+                    if value and not isinstance(value, _python_UUID):
+                        value = _python_UUID(value)
+                    return value
+
+                return process
+            else:
+
+                def process(value):
+                    if value and isinstance(value, _python_UUID):
+                        value = str(value)
+                    return value
+
+                return process
+        else:
+            if not self.as_uuid:
+
+                def process(value):
+                    if value and isinstance(value, _python_UUID):
+                        value = str(value)
+                    return value
+
+                return process
+            else:
+                return None
+
+
 class BIT(sqltypes.TypeEngine):
     __visit_name__ = "BIT"