时间戳

首页 > 解决方案 > Snowflake：Sqlalchemy ORM 更新变体字段

python - Snowflake：Sqlalchemy ORM 更新变体字段

问题描述

我在使用 sqlalchemy 更新表示雪花中的表的 python 模型时遇到问题。其中一列是变量字段，当我尝试更新模型并将其保存回数据库时，雪花抱怨该值不是变量。我尝试将数据作为字典和 json 字符串传递，但都不起作用。

模型属性定义为：

data = db.Column(snowsql.VARIANT)

snowsql 是 snowflake-sqlalchemy 软件包的别名，该软件包已安装，但似乎只提供类型方言，而没有提供太多其他功能。

有没有人有这样做的经验？

更新：

我添加的数据是一个 python 字典列表：

new_data['data'] = [{'component': None, 'display_name': 'blah', 'key': None, 'row': {'display_name': 'blah', 'type': 'blah'}, 'sub_type': None, 'type': 'blah'}]

设置变体属性时执行 json.dumps：

new_blah = Blah(
  data = json.dumps(new_data['data'])
)

结果是：

表达式类型与列数据类型不匹配，期望 VARIANT 但获得 VARCHAR(8905) 列 DATA

转储生成的 json：

"[{\"component\":null,\"display_name\":\"blah\",\"key\":null,\"row\":{\"display_name\":\"blah\",\"type\":\"blah\"},\"sub_type\":null,\"type\":\"blah\"}]"

或者，尝试在不执行 json.dumps 的情况下分配列表：

sqlalchemy.exc.ProgrammingError：（snowflake.connector.errors.ProgrammingError）252004：无：处理pyformat参数失败；'dict' 对象没有属性 'replace'

标签： pythonsqlalchemysnowflake-cloud-data-platformvariant

解决方案

据我所知，你正在尝试的目前是不可能的。我遇到了同样的问题，尽管我对SQLAlchemy 自定义类型有相当多的了解，但我目前无法解决它。

问题如下：

  1. 要插入VARIANT列，您需要使用 Snowflake 函数PARSE_JSON。
  2. 要使用所述功能，您需要在SELECT子句的上下文中，即您不能将其与INSERT ... VALUES格式一起使用（请参阅Snowflake Community）

我不知道有任何方法可以强制 SQLAlchemy 创建子查询，因为它是必需的。

万一这在未来发生变化，这是我走了多远：

class SnowflakeJSON(TypeDecorator):
    impl = VARIANT

    def process_bind_param(self, value, dialect):
        return json_serialize(value)

    def process_result_value(self, value, dialect):
        return json_deserialize(value)

    def bind_expression(self, bindparam):
        return func.PARSE_JSON(bindparam, type_=self)

    def copy(self, **kw):
        return SnowflakeJSON()

这会生成如下查询：

INSERT INTO my_table (key, data) 
VALUES ('001', PARSE_JSON('{"test": "hello world"}'));

但是，根据上面的雪花社区条目，该VALUES格式不起作用，您需要以下格式：

INSERT INTO my_table (key, data) 
SELECT '001', PARSE_JSON('{"test": "hello world"}');

作为后备，您始终可以使用将 JSON 对象存储为的列类型VARCHAR：

class SnowflakeJSON(TypeDecorator):
    impl = String

    def process_bind_param(self, value, dialect):
        return json_serialize(value)

    def process_result_value(self, value, dialect):
        return json_deserialize(value)

    def copy(self, **kw):
        return SnowflakeJSON()

这样一来，您显然会失去将 JSON 字符串存储为VARIANT数据类型的好处。但是，您可以在稍后阶段解析它，即使在查询中：

SELECT PARSE_JSON(text)['test'] from my_table;

推荐阅读