如何基于SQLAlchemy TypeDecorator实现通用TimeSeriesValue对象？

这个问题我之前在项目里也遇到过——要让时间序列值支持任意类型，又不想放弃SQLAlchemy的类型安全，确实有点棘手。我整理了几个实用的方案，你可以根据自己的需求选：

核心思路是让你的TimeSeriesValueType知道要转换到哪种Python类型，通过给TypeDecorator添加自定义参数来实现。这样每个TimeSeriesValue的value字段可以关联到对应TimeSeries指定的类型：

import json
from sqlalchemy.types import TypeDecorator, String

class TimeSeriesValueType(TypeDecorator):
    impl = String
    cache_ok = True  # 重要！告诉SQLAlchemy这个带参数的类型可以安全缓存

    def __init__(self, python_type, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.python_type = python_type

    def process_bind_param(self, value, dialect):
        # 针对不同类型做序列化：简单类型直接转字符串，复杂类型用JSON
        if value is None:
            return None
        if isinstance(value, (int, float, str, bool)):
            return str(value)
        # 自定义类型可以要求实现to_json方法，或者直接用json.dumps
        return json.dumps(value, default=lambda obj: obj.__dict__)

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        # 反向转换：简单类型直接强转，复杂类型用JSON加载后转目标类型
        if self.python_type in (int, float, bool):
            return self.python_type(value)
        elif self.python_type is str:
            return value
        # 自定义类型可以要求实现from_json方法
        try:
            raw_data = json.loads(value)
            if hasattr(self.python_type, 'from_json'):
                return self.python_type.from_json(raw_data)
            return self.python_type(**raw_data)
        except (json.JSONDecodeError, TypeError):
            # 兼容旧数据或者转换失败的情况
            return value

然后在定义TimeSeries和TimeSeriesValue的时候，通过TimeSeries的类型字段来动态指定value的类型：

from sqlalchemy import Column, Integer, DateTime, ForeignKey, String
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class TimeSeries(Base):
    __tablename__ = 'time_series'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # 存储Python类型的标识，比如"int"、"str"或者自定义类的路径"myapp.models.MyData"
    value_type_str = Column(String, nullable=False)

    @property
    def value_type(self):
        # 解析类型字符串为实际Python类型
        import importlib
        if '.' in self.value_type_str:
            module_name, class_name = self.value_type_str.rsplit('.', 1)
            module = importlib.import_module(module_name)
            return getattr(module, class_name)
        else:
            return eval(self.value_type_str)

class TimeSeriesValue(Base):
    __tablename__ = 'time_series_value'
    id = Column(Integer, primary_key=True)
    time_series_id = Column(Integer, ForeignKey('time_series.id'))
    timestamp = Column(DateTime, nullable=False)
    # 这里不能直接用TimeSeriesValueType，因为类型参数需要从关联的TimeSeries获取
    # 所以我们用属性封装来实现动态转换
    _value = Column(String, nullable=False)
    time_series = relationship("TimeSeries", back_populates="values")

    @property
    def value(self):
        # 从关联的TimeSeries获取目标类型，然后转换
        type_converter = TimeSeriesValueType(self.time_series.value_type)
        return type_converter.process_result_value(self._value, None)

    @value.setter
    def value(self, value):
        type_converter = TimeSeriesValueType(self.time_series.value_type)
        self._value = type_converter.process_bind_param(value, None)

TimeSeries.values = relationship("TimeSeriesValue", back_populates="time_series")

这个方案的优势是不需要修改数据库结构，就能支持几乎所有类型，缺点是查询时无法直接用value字段做数据库层面的类型过滤（比如数值大小比较）。

如果需要数据库层面的类型约束，比如数值型值存Float字段、字符串存String字段，可以用SQLAlchemy的多态继承来实现：

from sqlalchemy import Column, Integer, DateTime, Float, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

# 基类时间序列
class TimeSeries(Base):
    __tablename__ = 'time_series'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # 多态标识，区分不同类型的时间序列
    type = Column(String, nullable=False)
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'base'
    }
    values = relationship("TimeSeriesValue", back_populates="time_series")

# 数值型时间序列子类
class NumericTimeSeries(TimeSeries):
    __mapper_args__ = {
        'polymorphic_identity': 'numeric'
    }
    # 关联对应的数值型值子类
    values = relationship("NumericTimeSeriesValue", back_populates="time_series")

# 字符串型时间序列子类
class StringTimeSeries(TimeSeries):
    __mapper_args__ = {
        'polymorphic_identity': 'string'
    }
    values = relationship("StringTimeSeriesValue", back_populates="time_series")

# 基类时间序列值
class TimeSeriesValue(Base):
    __tablename__ = 'time_series_value'
    id = Column(Integer, primary_key=True)
    time_series_id = Column(Integer, ForeignKey('time_series.id'))
    timestamp = Column(DateTime, nullable=False)
    time_series = relationship("TimeSeries", back_populates="values")
    # 多态标识
    type = Column(String, nullable=False)
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'base'
    }

# 数值型值子类
class NumericTimeSeriesValue(TimeSeriesValue):
    __tablename__ = 'numeric_time_series_value'
    id = Column(Integer, ForeignKey('time_series_value.id'), primary_key=True)
    value = Column(Float, nullable=False)
    __mapper_args__ = {
        'polymorphic_identity': 'numeric'
    }

# 字符串型值子类
class StringTimeSeriesValue(TimeSeriesValue):
    __tablename__ = 'string_time_series_value'
    id = Column(Integer, ForeignKey('time_series_value.id'), primary_key=True)
    value = Column(String, nullable=False)
    __mapper_args__ = {
        'polymorphic_identity': 'string'
    }

这个方案的优势是类型完全安全，数据库里每个类型的值存在对应的字段，支持SQL层面的类型查询；缺点是新增类型需要创建新的子类和数据库表，扩展性稍弱。

如果追求最快的实现速度，可以直接用JSON字段存储值，然后在TimeSeries里记录类型，通过hybrid_property自动转换：

import json
from sqlalchemy import Column, Integer, DateTime, JSON, ForeignKey, String
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class TimeSeries(Base):
    __tablename__ = 'time_series'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    value_type_str = Column(String, nullable=False)

    def get_value_type(self):
        # 解析类型字符串
        import importlib
        if '.' in self.value_type_str:
            module_name, class_name = self.value_type_str.rsplit('.', 1)
            module = importlib.import_module(module_name)
            return getattr(module, class_name)
        else:
            return eval(self.value_type_str)

class TimeSeriesValue(Base):
    __tablename__ = 'time_series_value'
    id = Column(Integer, primary_key=True)
    time_series_id = Column(Integer, ForeignKey('time_series.id'))
    timestamp = Column(DateTime, nullable=False)
    _value = Column(JSON, nullable=False)
    time_series = relationship("TimeSeries", back_populates="values")

    @hybrid_property
    def value(self):
        if self._value is None:
            return None
        value_type = self.time_series.get_value_type()
        # 简单类型直接转换
        if value_type in (int, float, str, bool):
            return value_type(self._value)
        # 自定义类型支持from_dict方法
        if hasattr(value_type, 'from_dict'):
            return value_type.from_dict(self._value)
        return self._value

    @value.setter
    def value(self, value):
        if isinstance(value, (int, float, str, bool, type(None))):
            self._value = value
        elif hasattr(value, 'to_dict'):
            self._value = value.to_dict()
        else:
            # 默认转成JSON可序列化的结构
            self._value = json.loads(json.dumps(value, default=lambda obj: obj.__dict__))

TimeSeries.values = relationship("TimeSeriesValue", back_populates="time_series")

这个方案的优势是代码量少，不需要自定义TypeDecorator，缺点是JSON字段的查询性能不如原生类型，而且复杂类型的转换需要自己处理序列化逻辑。

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