如何在SQLAlchemy中处理SQL Server的SQL_VARIANT数据类型？

问题背景 #

需要通过Python获取SQL Server某张表的全部数据，但表中[fan_attribute.value]列为SQL_VARIANT类型，可存储datetime、int、float等多种数据类型。尝试用String、int、float、datetime等类型映射该列时，均触发如下错误：

sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('ODBC SQL type -150 is not yet supported. column-index=6 type=-150', 'HY106')

相关代码 #

SQL Server建表语句：

create table table_name(
[fan_attribute.value] sql_variant NULL
);

Python代码（原尝试）：

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base

from datetime import datetime
from sqlmodel import Field, PrimaryKeyConstraint, SQLModel

Base = declarative_base()

class ModelName(Base):
    __table_args__ = ({"schema": "can't disclose"})
    __tablename__ = "table_name"
    value = Column("fan_attribute.value", String)

解决方案 #

方法1：查询时将SQL_VARIANT转为字符串并按需转换 #

直接在SQL语句中把sql_variant列转换为字符串，同时可通过SQL_VARIANT_PROPERTY获取原始类型，再转换为对应Python类型：

from sqlalchemy import create_engine, text
from datetime import datetime

engine = create_engine("mssql+pyodbc://user:password@dsn_name")

with engine.connect() as conn:
    result = conn.execute(text("""
        SELECT 
            [fan_attribute.value] AS raw_value,
            SQL_VARIANT_PROPERTY([fan_attribute.value], 'BaseType') AS base_type
        FROM table_name
    """))
    for row in result:
        raw_val = row.raw_value
        base_type = row.base_type
        # 根据原始类型转换为对应Python类型
        if base_type == 'int':
            val = int(raw_val)
        elif base_type in ('datetime', 'smalldatetime'):
            val = raw_val  # pyodbc会自动转为datetime对象
        elif base_type in ('float', 'real'):
            val = float(raw_val)
        else:
            val = str(raw_val)
        print(val, type(val))

方法2：自定义SQLAlchemy类型处理SQL_VARIANT #

创建自定义类型，让SQLAlchemy能兼容sql_variant类型的解析：

from sqlalchemy.types import TypeDecorator, VARCHAR
import pyodbc
from sqlalchemy import create_engine, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

class SQLVariant(TypeDecorator):
    impl = VARCHAR
    cache_ok = True

    def process_result_value(self, value, dialect):
        # 返回pyodbc处理后的原始值，保留类型信息
        return value

Base = declarative_base()

# 修改Model定义
class ModelName(Base):
    __table_args__ = ({"schema": "can't disclose"})
    __tablename__ = "table_name"
    value = Column("fan_attribute.value", SQLVariant)

# 执行查询
engine = create_engine("mssql+pyodbc://user:password@dsn_name")
Session = sessionmaker(bind=engine)
session = Session()

rows = session.query(ModelName).all()
for row in rows:
    print(row.value, type(row.value))

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