sqlalchemy在Postgres中添加 /更新JSONB字段

Question

我正在尝试使用SQLalchemy在Postgres数据库中的JSONB字段中更新值。

一直在尝试使用func.jsonb_set，但我无法算出如何实现它。

使用表（test < / code>）的表（< / code>>），我将瞄准添加 /编辑JSON数据的通用方法。

ID	数据	名称
1	{“年龄”：44，“名称”：“ Barry”，“孩子”：[“ Baz”，“ Jim”]}	Barry
2	{“ Age”：47，“ Name”：“ Dave”，“”儿童”：[“ jeff”，“ jane”]}	dave

在Postgres中进行以下作品以进行简单更新。

UPDATE "test" SET "data"=jsonb_set("data"::jsonb, '{age}', '45')
WHERE "data"::json->>'name'='dave';

我可以使用Update来更新这样的单个值：

testobj_res.update(
    {
        TestObj.data: cast(
            cast(TestObj.data, JSONB).concat(func.jsonb_build_object("age", 45)),
            JSON,
        )
    }
)

session.commit()

我想 EG {“名称”：“ Barry”，“ Age”：45，“高度”的多个字段的更新：150}。

为了通过 .jsonb_set 添加更复杂的JSON结构而不是（'age'，45）的想法

testobj_res.first().data = func.jsonb_set(
    TestObj.data.cast(JSONB),
    ("age", 45),
    cast(TestObj.data, JSONB))

session.commit()

，但正在获取：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedFunction) function jsonb_set(jsonb, record, jsonb) does not exist
LINE 1: UPDATE public.test SET data=jsonb_set(CAST(public.test.data ...
                                    ^
HINT:  No function matches the given name and argument types. You might need to add explicit type casts.

完整示例代码：

import os
from sqlalchemy.dialects.postgresql import JSON, JSONB
from sqlalchemy import func, cast
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import urllib
from dotenv import load_dotenv

load_dotenv()

user = urllib.parse.quote_plus(os.environ.get("DB_USER"))
passwd = urllib.parse.quote_plus(os.environ.get("DB_PW"))

DB_URL = "postgresql://{}:{}@{}:{}/{}".format(
    user,
    passwd,
    os.environ.get("DB_HOST"),
    os.environ.get("DB_PORT"),
    os.environ.get("DB_NAME"),
)

engine = sa.create_engine(DB_URL)
Session = sessionmaker(bind=engine, autoflush=True)

session = Session()

Base = declarative_base()


class TestObj(Base):
    __tablename__ = "test"
    __table_args__ = {"autoload_with": engine, "schema": "public"}


testobj_res = session.query(TestObj).filter(TestObj.name == "dave")

testobj_res.first().data = func.jsonb_set(
    TestObj.data.cast(JSONB),
    ("age", 45),
    cast(TestObj.data, JSONB))

session.commit()

Answer 1

我有第二个参数并发出了问题，它铸造为varchar，不起作用。
对于sqlalchemy2.0此代码有效

func.jsonb_set(
        Model.jsonb_field,
        text("'{%s}'" % key),
        cast("value", JSONB),
    )

Answer 2

该代码

testobj_res.first().data = func.jsonb_set(
    TestObj.data.cast(JSONB),
    ("age", 45),
    cast(TestObj.data, JSONB))

正在通过JSONB对象，A record 代码> JSONB_SET ，但是该功能期望JSONB对象，JSONPATH字符串和该路径的新值，如JSONB ^*。

解开（'Age'，45）元组并删除最终JSONB将产生所需的结果。

with Session() as s, s.begin():
    testobj = s.scalar(sa.select(TestObj).limit(1))
    testobj.data = sa.func.jsonb_set(
        sa.cast(testobj.data, postgresql.JSONB),
        '{age}',
        sa.cast(42, postgresql.JSONB)
    )

---

^*还有一个可选的布尔create-if-not-not参数，我们可以忽略。