I have a custom type for interfacing with INET6 in MariaDB, and from process_bind_param
I want to return an SQL expression like this: CAST('2001::ff' AS INET6)
I’ve tried this so far (slightly simplified):
import ipaddress
from sqlalchemy import types
from sqlalchemy.sql.expression import cast
class Inet6(types.TypeDecorator):
impl = types.BINARY
cache_ok = True
def get_col_spec(self, **kw):
return 'INET6'
def process_bind_param(self, value, dialect):
if value is None:
return None
assert isinstance(value, ipaddress.IPv6Address)
return cast(str(value), self)
def process_result_value(self, value, dialect):
if value is None:
return None
return ipaddress.IPv6Address(value)
However, when I use it in a query, I get this exception:
Traceback (most recent call last):
File "/home/mlenz/nnis/nnis-business-services/ip-inet6.py", line 19, in <module>
for i in q.all():
^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/orm/query.py", line 2693, in all
return self._iter().all() # type: ignore
^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/orm/query.py", line 2847, in _iter
result: Union[ScalarResult[_T], Result[_T]] = self.session.execute(
^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 2308, in execute
return self._execute_internal(
^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 2190, in _execute_internal
result: Result[Any] = compile_state_cls.orm_execute_statement(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/orm/context.py", line 293, in orm_execute_statement
result = conn.execute(
^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1416, in execute
return meth(
^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 516, in _execute_on_connection
return connection._execute_clauseelement(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1639, in _execute_clauseelement
ret = self._execute_context(
^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1820, in _execute_context
self._handle_dbapi_exception(
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2343, in _handle_dbapi_exception
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1814, in _execute_context
context = constructor(
^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 1487, in _init_compiled
d_param = {
^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 1488, in <dictcomp>
key: flattened_processors[key](compiled_params[key])
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/type_api.py", line 2059, in process
return fixed_impl_processor(
^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/sqltypes.py", line 929, in process
return DBAPIBinary(value)
^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/pymysql/__init__.py", line 128, in Binary
return bytes(x)
^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/operators.py", line 666, in __getitem__
return self.operate(getitem, index)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 1616, in operate
return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/operators.py", line 666, in __getitem__
return self.operate(getitem, index)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/type_api.py", line 1733, in operate
return super().operate(op, *other, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/type_api.py", line 194, in operate
return op_fn(self.expr, op, *other, **addtl_kw)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/default_comparator.py", line 244, in _getitem_impl
_unsupported_impl(expr, op, other, **kw)
File "/home/mlenz/venv/lib/python3.11/site-packages/sqlalchemy/sql/default_comparator.py", line 250, in _unsupported_impl
raise NotImplementedError(
sqlalchemy.exc.StatementError: (builtins.NotImplementedError) Operator 'getitem' is not supported on this expression
[SQL: SELECT inet6_test.ip AS inet6_test_ip, inet6_test.id AS inet6_test_id, inet6_test.timestamp AS inet6_test_timestamp
FROM inet6_test
WHERE inet6_test.ip = %(ip_1)s
LIMIT %(param_1)s]
[parameters: [{}]]
Is this (returning a CAST
from process_bind_param
) supported at all? Or am I just doing it wrong?
(The reason I want to produce a CAST
is that because binary strings are far less readable in the emitted SQL/bind params, and for <
and >
comparisons, MariaDB needs explicit CAST
statements when dealing string representations of IP addresses. If I simply return a binary the_ip_address.packed
I get hard-to-debug errors where mariadb complains that the value isn’t a valid INET6)