Snowflake SQLAlchemy动态创建带默认值Timestamp列遇序列化错误

问题核心是JSON无法序列化SQLAlchemy的TextClause对象——你在JSON Schema里直接传入text('current_timestamp')是错误的，JSON仅支持存储字符串、数字等基础类型，无法存储Python对象。

1. 修改JSON Schema，将serverdefault改为字符串形式的SQL表达式 #

把原Schema中的text('current_timestamp')替换成字符串'current_timestamp()'：

json_cls_schema = {
    "clsname": "MyClass",
    "tablename": "my_table",
    "columns": [
        {"name": "id", "type": "integer", "is_pk": True, "is_auto" : True},
        {"name": "my_time_stamp", "type": "timestamp", 'serverdefault': 'current_timestamp()'}
    ],
}

2. 修改mapping_for_json函数，动态转换字符串为TextClause #

在生成Column对象时，判断如果serverdefault存在且为字符串，就用text()包装它：

from sqlalchemy import Column, Integer, TIMESTAMP, text
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

_type_lookup = {
    "integer": Integer,
    "timestamp": TIMESTAMP,
}

def mapping_for_json(json_cls_schema):
    clsdict = {"__tablename__": json_cls_schema["tablename"]}

    clsdict.update(
        {
            rec["name"]: Column(
                _type_lookup[rec["type"]],
                primary_key=rec.get("is_pk", False),
                autoincrement=rec.get("is_auto", False),
                serverdefault=text(rec["serverdefault"]) if rec.get("serverdefault") else None
            )
            for rec in json_cls_schema["columns"]
        }
    )

    return type(json_cls_schema["clsname"], (Base,), clsdict)

关键说明 #

• JSON Schema仅存储字符串，彻底避免序列化错误；
• 动态生成Column时，通过text()将字符串形式的SQL表达式转换为SQLAlchemy可识别的TextClause对象，完全匹配你期望的server_default=text('current_timestamp()')效果；
• 将原默认值''改为None，符合Column参数规范（None表示无默认值，空字符串可能引发异常）。

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