非PostgreSQL环境下SQLAlchemy自定义向量列实现方案求助

已知此前有类似问题，但未找到符合需求的解答，若有重复深表歉意，请标记为重复问题！

我正在用SQLAlchemy处理包含笛卡尔坐标的数据库，现有代码需要手动处理N维向量（如3D坐标）的存储、查询和对象构造，存在大量重复工作，希望封装成通用的Vector列类型来简化操作。

现有代码与痛点 #

1. 资产类定义 #

当前需要为向量的每个维度单独声明字段，再手动添加属性封装：

import sqlalchemy as sql

class Asset(Base):
    """可渲染的3D对象"""
    __tablename__ = "assets"

    id = sql.Column(sql.Integer, primary_key=True)
    mesh = sql.Column(sql.String(80), nullable=False)
    
    # 手动声明3个维度字段
    position_x = sql.Column(sql.Float, nullable=False)
    position_y = sql.Column(sql.Float, nullable=False)
    position_z = sql.Column(sql.Float, nullable=False)

    @property
    def position(self):
        return self.position_x, self.position_y, self.position_z

2. 添加资产的去重逻辑 #

查询时需要为每个维度编写重复的容差判断代码，构造对象时还要逐个赋值：

from typing import Tuple

def add_asset(mesh: str, position: Tuple[float, float, float]) -> bool:
    engine = sql.create_engine("DB_URL")
    with sql.orm.Session(engine) as session:
        x, y, z = position
        # 重复编写3个维度的容差判断
        asset = session.query(Asset).filter(
            sql.func.abs(x - Asset.position_x) < 1e-6,
            sql.func.abs(y - Asset.position_y) < 1e-6,
            sql.func.abs(z - Asset.position_z) < 1e-6
        ).filter_by(mesh=mesh).first()

        if asset is not None:
            return False

        # 手动逐个维度赋值
        session.add(Asset(mesh=mesh, position_x=x, position_y=y, position_z=z))
        session.commit()
        return True

3. 期望的理想状态 #

希望通过Vector类型简化所有操作，代码如下：

class Asset(Base):
    __tablename__ = "assets"

    id = sql.Column(sql.Integer, primary_key=True)
    mesh = sql.Column(sql.String(80), nullable=False)
    # 直接使用Vector(3)定义3D向量
    position = sql.Column(Vector(3), nullable=False)

def add_asset(mesh: str, position: Tuple[float, float, float]) -> bool:
    engine = sql.create_engine("DB_URL")
    with sql.orm.Session(engine) as session:
        # 用统一的close函数处理容差查询
        asset = session.query(Asset).filter(
            close(position, Asset.position, 1e-6)
        ).filter_by(mesh=mesh).first()

        if asset is not None:
            return False

        # 直接传入元组构造对象
        session.add(Asset(mesh=mesh, position=position))
        session.commit()
        return True

注：无法使用PostgreSQL的ARRAY类型，需要兼容其他数据库（如SQLite）。

通过SQLAlchemy的CompositeType封装通用向量类型，自动处理字段生成、属性访问和查询逻辑，完全消除重复代码。

1. 实现Vector复合类型 #

import sqlalchemy as sql
from sqlalchemy import types
from sqlalchemy.ext.compiler import compiles
from typing import Tuple, TypeVar, Generic

T = TypeVar('T', bound=float)

class Vector(types.CompositeType, Generic[T]):
    def __init__(self, dimension: int, col_type: types.TypeEngine = types.Float, nullable: bool = False):
        self.dimension = dimension
        self.col_type = col_type
        # 生成维度字段名：如3D对应x、y、z
        fields = [chr(ord('x') + i) for i in range(dimension)]
        super().__init__(f"vector_{dimension}", fields, [col_type]*dimension, nullable=nullable)

    class comparator_factory(types.CompositeType.comparator_factory):
        def close(self, other: Tuple[T, ...], tolerance: float = 1e-6):
            """生成向量各维度差值小于容差的查询条件"""
            clauses = []
            for field, val in zip(self.type.fields, other):
                col = getattr(self.expr, field)
                clauses.append(sql.func.abs(col - val) < tolerance)
            return sql.and_(*clauses)

# 为不支持原生复合类型的数据库（如SQLite）做兼容处理
@compiles(Vector, 'sqlite')
def compile_vector_sqlite(type_, compiler, **kw):
    return ", ".join([f"{field} {compiler.process(type_.col_type)}" for field in type_.fields])

2. 改造Asset类 #

直接使用Vector类型定义向量字段，无需手动声明维度列：

class Asset(Base):
    """可渲染的3D对象"""
    __tablename__ = "assets"

    id = sql.Column(sql.Integer, primary_key=True)
    mesh = sql.Column(sql.String(80), nullable=False)
    # 3D向量字段
    position = sql.Column(Vector(3), nullable=False)

3. 简化添加资产的逻辑 #

直接调用向量列的close方法做容差查询，构造对象时传入元组即可：

def add_asset(mesh: str, position: Tuple[float, float, float]) -> bool:
    engine = sql.create_engine("DB_URL")
    with sql.orm.Session(engine) as session:
        asset = session.query(Asset).filter(
            Asset.position.close(position, 1e-6)
        ).filter_by(mesh=mesh).first()

        if asset is not None:
            return False

        session.add(Asset(mesh=mesh, position=position))
        session.commit()
        return True

• 支持任意提前指定的维度（2D、3D、6D等），只需修改Vector的参数即可
• 自动生成对应维度的数据库列，无需手动声明
• 构造对象时直接传入元组，SQLAlchemy自动拆分赋值
• 查询时通过close方法统一处理容差判断，无需重复编写维度逻辑
• 兼容SQLite等不支持原生复合类型的数据库，自动降级为单独列存储

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