Support VARBINARY for uuids (#301)

.flake8

@@ -7,4 +7,4 @@ select =
     W,  # flake8-pycodestyle
 ignore=
     # conflict with black formatter
-    W503,E203,
+    W503,E203,E704,

CHANGES.rst

@@ -9,6 +9,8 @@ Features
 
 - ``sqlalchemy_hana.errors`` will now raise a ``SequenceLockTimeoutError`` error for error
   messages a lock wait timeout error caused by a sequence
+- uuid types can now be backed by ``VARBINARY`` instead of ``NVARCHAR``. For this use the
+  ``sqlalchemy_hana.types.Uuid`` type with ``as_varbinary=True``
 
 2.4.0
 -----

README.rst

@@ -180,7 +180,7 @@ The following table shows the mapping:
     * - DOUBLE_PRECISION
       - DOUBLE
     * - Uuid
-      - NVARCHAR(32)
+      - NVARCHAR(32) / VARBINARY(16)
     * - LargeBinary
       - BLOB
     * - UnicodeText
@@ -198,6 +198,18 @@ The ``ARRAY`` datatype is not supported because ``hdbcli`` does not yet provide
 The ``JSON`` datatype only supports saving/updating field contents, but no json-based filters/deep indexing,
 as these are not supported by SAP HANA.
 
+The ``Uuid`` (note the casing) supports a special flag ``as_varbinary``.
+If set to true (by default false), the UUID will be stored as a ``VARBINARY(16)`` instead of a ``NVARCHAR(32)``.
+This does not effect the python side, meaning depending on the ``as_uuid`` flag, either uuid
+objects or strings are used.
+To use this feature in a database agnostic way, use
+``UuidType = Uuid.with_variant(sqlalchemy_hana.types.Uuid(as_varbinary=True), "hana")``.
+Note, that SAP HANA offers two UUID functions
+(`NEWUID <https://help.sap.com/docs/hana-cloud-database/sap-hana-cloud-sap-hana-database-sql-reference-guide/newuid-function-miscellaneous?locale=en-US>`_
+and `SYSUUID <https://help.sap.com/docs/hana-cloud-database/sap-hana-cloud-sap-hana-database-sql-reference-guide/sysuuid-function-miscellaneous?locale=en-US>`_
+) which can be used to generate e.g. default values like
+``Column('id', Uuid, server_default=func.NEWUID)``.
+
 Regex
 ~~~~~
 sqlalchemy-hana supports the ``regexp_match`` and ``regexp_replace``

sqlalchemy_hana/_uuid.py

@@ -0,0 +1,50 @@
+"""UUID type."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, TypeVar
+from uuid import UUID as PyUUID
+
+from sqlalchemy import types as sqltypes
+from sqlalchemy.engine import Dialect
+
+_RET = TypeVar("_RET", str, PyUUID)
+
+
+class Uuid(sqltypes.Uuid[_RET]):
+
+    def __init__(
+        self,
+        as_uuid: bool = True,
+        native_uuid: bool = True,
+        as_varbinary: bool = False,
+    ) -> None:
+        super().__init__(as_uuid, native_uuid)  # type:ignore
+        self.as_varbinary = as_varbinary
+
+    def bind_processor(self, dialect: Dialect) -> Callable[[Any | None], Any | None]:
+        if not self.as_varbinary:
+            return super().bind_processor(dialect)
+
+        def process(value: Any | None) -> Any | None:
+            if value is None:
+                return value
+            uuid = value if isinstance(value, PyUUID) else PyUUID(value)
+            return uuid.bytes
+
+        return process
+
+    def result_processor(
+        self, dialect: Dialect, coltype: Any
+    ) -> Callable[[Any | None], Any | None]:
+        if not self.as_varbinary:
+            return super().result_processor(dialect, coltype)
+
+        def process(value: Any | None) -> Any | None:
+            if value is None:
+                return value
+            if self.as_uuid:
+                return PyUUID(bytes=value.tobytes())
+            return str(PyUUID(bytes=value.tobytes()))
+
+        return process

sqlalchemy_hana/dialect.py

@@ -375,7 +375,8 @@ def visit_DOUBLE_PRECISION(self, type_: types.TypeEngine[Any], **kw: Any) -> str
         return super().visit_DOUBLE(type_, **kw)
 
     def visit_uuid(self, type_: types.TypeEngine[Any], **kw: Any) -> str:
-        # SAP HANA has no UUID type, therefore delegate to NVARCHAR(32)
+        if isinstance(type_, hana_types.Uuid) and type_.as_varbinary:
+            return "VARBINARY(16)"
         return self._render_string_type(type_, "NVARCHAR", length_override=32)
 
     def visit_JSON(self, type_: types.TypeEngine[Any], **kw: Any) -> str:
@@ -521,6 +522,7 @@ class HANAHDBCLIDialect(default.DefaultDialect):
     supports_statement_cache = True
     supports_unicode_binds = True
     supports_unicode_statements = True
+    supports_native_uuid = False
     support_views = True
 
     colspecs = {
@@ -531,6 +533,9 @@ class HANAHDBCLIDialect(default.DefaultDialect):
         # the wrong class will be used
         hana_types.SECONDDATE: hana_types.SECONDDATE,
     }
+    if sqlalchemy.__version__ >= "2":
+        colspecs[types.Uuid] = hana_types.Uuid
+
     types_with_length = [
         hana_types.VARCHAR,
         hana_types.NVARCHAR,

sqlalchemy_hana/types.py

@@ -148,7 +148,7 @@ class JSON(sqltypes.JSON):
     pass
 
 
-__all__ = (
+__all__ = [
     "ALPHANUM",
     "BIGINT",
     "BLOB",
@@ -174,4 +174,11 @@ class JSON(sqltypes.JSON):
     "TINYINT",
     "VARBINARY",
     "VARCHAR",
-)
+]
+
+
+if sqlalchemy.__version__ >= "2":
+    # pylint: disable=unused-import
+    from sqlalchemy_hana._uuid import Uuid  # noqa: F401
+
+    __all__.append("Uuid")

test/test_types.py

@@ -7,6 +7,7 @@
 import decimal
 import random
 from unittest import mock
+from uuid import UUID
 
 import pytest
 import sqlalchemy
@@ -229,3 +230,38 @@ def test_compile(self, connection, metadata) -> None:
             str(CreateTable(mytab).compile(connection))
             == "\nCREATE TABLE mytab (\n\tmycol ALPHANUM(10)\n)\n\n"
         )
+
+
+if sqlalchemy.__version__ >= "2":
+
+    class StringUUIDAsStringTest(_TypeBaseTest):
+        column_type = hana_types.Uuid(as_uuid=False)
+        data = "9f01b2fb-bf0d-4b46-873c-15d0976b4100"
+
+        @property
+        def reflected_column_type(self):
+            return hana_types.NVARCHAR(length=32)
+
+    class StringUUIDAsUUIDTest(_TypeBaseTest):
+        column_type = hana_types.Uuid(as_uuid=True)
+        data = UUID("9f01b2fb-bf0d-4b46-873c-15d0976b4100")
+
+        @property
+        def reflected_column_type(self):
+            return hana_types.NVARCHAR(length=32)
+
+    class BinaryUUIDAsStringTest(_TypeBaseTest):
+        column_type = hana_types.Uuid(as_uuid=False, as_varbinary=True)
+        data = "9f01b2fb-bf0d-4b46-873c-15d0976b4100"
+
+        @property
+        def reflected_column_type(self):
+            return hana_types.VARBINARY(length=16)
+
+    class BinaryUUIDAsUUIDTest(_TypeBaseTest):
+        column_type = hana_types.Uuid(as_uuid=True, as_varbinary=True)
+        data = UUID("9f01b2fb-bf0d-4b46-873c-15d0976b4100")
+
+        @property
+        def reflected_column_type(self):
+            return hana_types.VARBINARY(length=16)