SQLAlchemy操作SQLite3：批量移除JSON列指定ID并清理空行

假设你的模型定义大致如下：

from sqlalchemy import Column, Integer, JSON
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class MyModel(Base):
    __tablename__ = "my_table"
    id = Column(Integer, primary_key=True)
    ids = Column(JSON, nullable=False)

1. 批量查询包含指定ID的行 #

直接利用数据库的JSON数组包含判断，以PostgreSQL为例（其他数据库可对应替换函数）：

from sqlalchemy import func

target_id = 123
# 查询所有ids数组中包含target_id的行
query = session.query(MyModel).filter(func.jsonb_contains(MyModel.ids, func.to_jsonb(target_id)))

2. 批量移除指定ID并处理空列表删除 #

最优雅的方式是用**CTE（公共表表达式）**完成「更新+删除」的原子操作，避免多次查询和逐行迭代：

from sqlalchemy import update, delete, text

target_id = 123

# 第一步：定义CTE，先更新符合条件的行，移除指定ID
update_cte = (
    update(MyModel)
    .where(func.jsonb_contains(MyModel.ids, func.to_jsonb(target_id)))
    .values(ids=func.jsonb_array_remove(MyModel.ids, func.to_jsonb(target_id)))
    .returning(MyModel.id, MyModel.ids)
).cte("updated_rows")

# 第二步：删除更新后ids为空数组的行
delete_stmt = delete(MyModel).where(
    MyModel.id.in_(update_cte.c.id),
    update_cte.c.ids == text("'[]'::jsonb")
)

# 执行操作
session.execute(delete_stmt)
session.commit()

说明 #

• 上述代码基于PostgreSQL的jsonb类型操作（如果你的列是JSON而非JSONB，可以把jsonb_contains、jsonb_array_remove换成json_array_elements相关判断，或者先转成jsonb处理）。
• 如果是MySQL数据库，替换对应的JSON函数即可：比如用JSON_CONTAINS判断包含，用JSON_REMOVE结合JSON_SEARCH来移除元素（注意MySQL的JSON_REMOVE需要指定路径，写法会稍复杂）。
• 整个操作是原子性的，通过CTE把更新和删除绑定在一起，避免逐行处理的性能问题。

内容的提问来源于stack exchange，提问作者Omri. B