如何在SQLAlchemy自定义类型的process_bind_param中返回函数表达式？

我为MariaDB的INET6类型实现了自定义TypeDecorator，希望在process_bind_param方法中返回类似CAST('2001::ff' AS INET6)的SQL表达式，但遇到了问题。

尝试的简化代码 #

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)

遇到的错误 #

使用该类型执行查询时，抛出NotImplementedError，提示操作符'getitem'不被支持。

疑问 #

• 从process_bind_param返回CAST表达式是否被支持？还是我的实现方式有误？
• 补充：是不是因为自定义类型以BINARY为基础类型，SQLAlchemy会把CAST语句转为binary()？如果是，能否禁止对绑定参数执行此转换？

二进制字符串问题补充 #

为说明直接传二进制的问题，我创建了如下表：

Create Table: CREATE TABLE `inet6_test` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `ip` inet6 NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=214872 DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_unicode_ci

执行以下代码时：

address = IPv6Address('2001::ff')
result = session.execute(
    text('INSERT INTO inet6_test SET ip=:ip'),
    {'ip': address.packed},
)

出现OperationalError，提示“INET6值不正确”。

但我确认IPv6Address.packed是正确格式：执行以下查询返回的十六进制值正确，且在MariaDB命令行执行SELECT CAST(X'200100000000000000000000000000FF' AS INET6)能得到正确的2001::ff，说明MariaDB能识别该输入，只是通过SQLAlchemy传递时出了问题。

涉及版本 #

• sqlalchemy==2.0.29
• pymysql==1.1.0
• mariadb 10.11.6

解决方案 #

问题根源 #

1. process_bind_param的返回值只能是原始参数值，不能返回SQL表达式（比如cast对象），这是SQLAlchemy的设计规则——TypeDecorator的绑定处理方法负责转换值，而非生成SQL片段，返回表达式会导致后续处理逻辑报错。
2. 直接传packed二进制时出错，是因为SQLAlchemy会把BINARY类型的值按普通二进制参数传递，而MariaDB的INET6类型对字节序或传递格式有特定要求，默认的BINARY绑定处理未适配该要求。

正确实现方式 #

方案1：用bind_expression生成CAST表达式（推荐）

SQLAlchemy 2.0+提供bind_expression方法，专门用于为绑定参数生成SQL转换逻辑，这是实现你需求的正确方式：

import ipaddress
from sqlalchemy import types
from sqlalchemy.sql.expression import cast

class Inet6(types.TypeDecorator):
    impl = types.VARCHAR(39)  # IPv6字符串最长39位
    cache_ok = True

    def get_col_spec(self, **kw):
        return 'INET6'

    def bind_expression(self, bindvalue):
        # 为绑定的字符串值生成CAST表达式
        return cast(bindvalue, self)

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        assert isinstance(value, ipaddress.IPv6Address)
        return str(value)

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return ipaddress.IPv6Address(value)

方案2：修复二进制传递问题

如果坚持用二进制传递，可通过literal_binary生成MariaDB能识别的二进制字面量：

import ipaddress
from sqlalchemy import types
from sqlalchemy.sql.expression import literal_binary

class Inet6(types.TypeDecorator):
    impl = types.BINARY(16)  # IPv6固定16字节
    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 literal_binary(value.packed)

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return ipaddress.IPv6Address(value)

关键说明 #

• bind_expression是官方推荐的生成绑定参数转换逻辑的方法，既满足生成CAST表达式的需求，也符合SQLAlchemy的设计规范。
• 字符串传递+CAST的方式，既能保证SQL的可读性，也能让MariaDB正确处理IP地址的比较操作（<、>等）。
• literal_binary会直接生成X'...'格式的SQL字面量，而非普通绑定参数，能被MariaDB正确识别为INET6格式的二进制数据。

内容的提问来源于stack exchange，提问作者moritz