Types: Add support for BINARY columns

CHANGES.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## Unreleased
+- Types: Added support for BLOB type, per base64 encoding
+
 ## 2026/06/17 0.43.0
 - Types: Improved support for FLOAT type, converging to FLOAT vs. DOUBLE
 - Types: Added method `ObjectArray.as_generic` for better reverse type lookups

src/sqlalchemy_cratedb/compiler.py

@@ -271,6 +271,9 @@ def visit_TIMESTAMP(self, type_, **kw):
         """
         return "TIMESTAMP %s" % ((type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",)
 
+    def visit_BLOB(self, type_, **kw):
+        return "STRING"
+
     def visit_FLOAT(self, type_, **kw):
         """
         From `sqlalchemy.sql.sqltypes.Float`.

src/sqlalchemy_cratedb/dialect.py

@@ -36,6 +36,7 @@
 )
 from .sa_version import SA_1_4, SA_2_0, SA_VERSION
 from .type import FloatVector, ObjectArray, ObjectType
+from .type.binary import LargeBinary
 from .util import SSLMode
 
 # For SQLAlchemy >= 1.1.
@@ -171,6 +172,7 @@ def process(value):
     sqltypes.Date: Date,
     sqltypes.DateTime: DateTime,
     sqltypes.TIMESTAMP: DateTime,
+    sqltypes.LargeBinary: LargeBinary,
 }
 
 if SA_VERSION >= SA_2_0:

src/sqlalchemy_cratedb/type/__init__.py

@@ -1,11 +1,13 @@
 from .array import ObjectArray
+from .binary import LargeBinary
 from .geo import Geopoint, Geoshape
 from .object import ObjectType
 from .vector import FloatVector, knn_match
 
 __all__ = [
     Geopoint,
     Geoshape,
+    LargeBinary,
     ObjectArray,
     ObjectType,
     FloatVector,

src/sqlalchemy_cratedb/type/binary.py

@@ -0,0 +1,35 @@
+import base64
+
+from sqlalchemy import String
+
+
+class LargeBinary(String):
+    """A type for large binary byte data.
+
+    The :class:`.LargeBinary` type corresponds to a large and/or unlengthed
+    binary type for the target platform, such as BLOB on MySQL and BYTEA for
+    PostgreSQL.  It also handles the necessary conversions for the DBAPI.
+
+    """
+
+    __visit_name__ = "large_binary"
+
+    def bind_processor(self, dialect):
+        if dialect.dbapi is None:
+            return None
+
+        def process(value):
+            if value is not None:
+                return base64.b64encode(value).decode()
+            else:
+                return None
+
+        return process
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if value is not None:
+                return base64.b64decode(value)
+            return value
+
+        return process

tests/test_type_binary.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import base64
+from unittest import TestCase
+from unittest.mock import MagicMock
+
+from sqlalchemy_cratedb.type.binary import LargeBinary
+
+
+class LargeBinaryBindProcessorTest(TestCase):
+    def setUp(self):
+        self.type = LargeBinary()
+        self.dialect = MagicMock()
+        self.dialect.dbapi = MagicMock()
+
+    def test_encodes_bytes_to_base64_string(self):
+        process = self.type.bind_processor(self.dialect)
+        result = process(b"hello world")
+        self.assertEqual(result, base64.b64encode(b"hello world").decode())
+
+    def test_returns_none_for_none_input(self):
+        process = self.type.bind_processor(self.dialect)
+        self.assertIsNone(process(None))
+
+    def test_returns_none_processor_when_dbapi_is_none(self):
+        self.dialect.dbapi = None
+        processor = self.type.bind_processor(self.dialect)
+        self.assertIsNone(processor)
+
+    def test_encodes_arbitrary_binary_data(self):
+        process = self.type.bind_processor(self.dialect)
+        data = bytes(range(256))
+        result = process(data)
+        self.assertEqual(result, base64.b64encode(data).decode())
+
+
+class LargeBinaryResultProcessorTest(TestCase):
+    def setUp(self):
+        self.type = LargeBinary()
+        self.dialect = MagicMock()
+
+    def test_decodes_base64_string_to_bytes(self):
+        process = self.type.result_processor(self.dialect, None)
+        encoded = base64.b64encode(b"hello world").decode()
+        result = process(encoded)
+        self.assertEqual(result, b"hello world")
+
+    def test_returns_none_for_none_input(self):
+        process = self.type.result_processor(self.dialect, None)
+        self.assertIsNone(process(None))
+
+    def test_round_trip(self):
+        bind = self.type.bind_processor(self.dialect)
+        result = self.type.result_processor(self.dialect, None)
+        data = b"\x00\x01\x02\xff\xfe\xfd"
+        self.assertEqual(result(bind(data)), data)