SQLite正确插入带标签条目及多表关联映射的实现方案咨询

你的需求完全可以实现，当前的三表结构是多对多关联的标准设计，符合第三范式要求，单SQL原子操作的方案可用，不需要拆分到业务代码处理6步流程。

你现有的表设计已经足够合理，唯一可以补充的是给关联表的外键加索引，提升后续关联查询的性能：

CREATE INDEX idx_tagmap_entry ON TagMap(entry);
CREATE INDEX idx_tagmap_tag ON TagMap(tag);

单条条目插入 #

适用于每次插入单条带标签的条目，操作全程原子，出错自动回滚：

WITH
-- 插入条目，已存在则直接返回ID
insert_entry AS (
    INSERT INTO Entry (name)
    VALUES ('待插入的条目名称')
    ON CONFLICT(name) DO UPDATE SET name = name
    RETURNING id AS entry_id
),
-- 插入不存在的标签，已存在则直接返回ID
insert_tags AS (
    INSERT INTO Tags (tag)
    VALUES ('标签1'), ('标签2'), ('标签3')
    ON CONFLICT(tag) DO UPDATE SET tag = tag
    RETURNING id AS tag_id, tag
),
-- 拉取所有需要的标签ID（兼容已存在的标签）
all_required_tags AS (
    SELECT id AS tag_id, tag FROM Tags WHERE tag IN ('标签1', '标签2', '标签3')
)
-- 插入关联关系，重复关联自动忽略
INSERT INTO TagMap (entry, tag)
SELECT ie.entry_id, art.tag_id
FROM insert_entry ie, all_required_tags art
ON CONFLICT(entry, tag) DO NOTHING;

批量JSON数据导入 #

如果需要一次性导入整个JSON数组，可直接用SQLite自带的JSON函数处理，不需要在脚本层解析：

WITH
-- 解析输入的JSON数组
input_data AS (
    SELECT
        json_extract(value, '$.name') AS entry_name,
        json_extract(value, '$.tags') AS tag_list
    FROM json_each(@输入的JSON字符串参数)
),
-- 批量插入所有条目
batch_insert_entries AS (
    INSERT INTO Entry (name)
    SELECT entry_name FROM input_data
    ON CONFLICT(name) DO UPDATE SET name = name
    RETURNING id AS entry_id, name AS entry_name
),
-- 展开所有去重后的标签
all_distinct_tags AS (
    SELECT DISTINCT json_extract(json_each.value, '$') AS tag
    FROM input_data, json_each(input_data.tag_list)
),
-- 批量插入不存在的标签
batch_insert_tags AS (
    INSERT INTO Tags (tag)
    SELECT tag FROM all_distinct_tags
    ON CONFLICT(tag) DO UPDATE SET tag = tag
    RETURNING id AS tag_id, tag
),
-- 拉取所有需要的标签ID
all_tag_ids AS (
    SELECT id AS tag_id, tag FROM Tags WHERE tag IN (SELECT tag FROM all_distinct_tags)
),
-- 组装条目和标签的ID对
entry_tag_pairs AS (
    SELECT
        bie.entry_id,
        ati.tag_id
    FROM input_data id
    JOIN batch_insert_entries bie ON id.entry_name = bie.entry_name
    JOIN json_each(id.tag_list) jt ON 1=1
    JOIN all_tag_ids ati ON json_extract(jt.value, '$') = ati.tag
)
-- 批量插入关联关系
INSERT INTO TagMap (entry, tag)
SELECT entry_id, tag_id FROM entry_tag_pairs
ON CONFLICT(entry, tag) DO NOTHING;

首先定义表模型：

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, create_engine
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy.dialects.sqlite import insert

Base = declarative_base()

class Entry(Base):
    __tablename__ = "Entry"
    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String, unique=True, nullable=False)
    # 多对多关联，自动映射TagMap中间表
    tags = relationship("Tag", secondary="TagMap", back_populates="entries")

class Tag(Base):
    __tablename__ = "Tags"
    id = Column(Integer, primary_key=True, nullable=False)
    tag = Column(String, unique=True, nullable=False)
    entries = relationship("Entry", secondary="TagMap", back_populates="tags")

class TagMap(Base):
    __tablename__ = "TagMap"
    entry = Column(Integer, ForeignKey("Entry.id"), nullable=False, primary_key=True)
    tag = Column(Integer, ForeignKey("Tags.id"), nullable=False, primary_key=True)
    __table_args__ = (UniqueConstraint("entry", "tag"),)

# 初始化连接
engine = create_engine("sqlite:///your_database.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

单条插入实现 #

def add_entry_with_tags(entry_name: str, tags: list[str]) -> int:
    with Session() as session:
        # 插入/获取条目ID
        entry_stmt = insert(Entry).values(name=entry_name).on_conflict_do_update(
            index_elements=["name"], set_={"name": entry_name}
        ).returning(Entry.id)
        entry_id = session.execute(entry_stmt).scalar_one()

        # 批量插入不存在的标签
        tag_stmt = insert(Tag).values([{"tag": t} for t in tags]).on_conflict_do_nothing()
        session.execute(tag_stmt)

        # 查询所有需要的标签ID
        tag_ids = session.query(Tag.id).filter(Tag.tag.in_(tags)).all()

        # 插入关联关系
        map_stmt = insert(TagMap).values(
            [{"entry": entry_id, "tag": t_id[0]} for t_id in tag_ids]
        ).on_conflict_do_nothing()
        session.execute(map_stmt)

        session.commit()
    return entry_id

ORM 简化写法 #

如果不需要严格的单SQL执行，用ORM的关联属性自动处理中间表，代码更简洁：

def add_entry_with_tags_orm(entry_name: str, tags: list[str]) -> int:
    with Session() as session:
        tag_objs = []
        for tag_name in tags:
            # 不存在则创建标签
            stmt = insert(Tag).values(tag=tag_name).on_conflict_do_nothing()
            session.execute(stmt)
            # 查询标签对象
            tag = session.query(Tag).filter(Tag.tag == tag_name).one()
            tag_objs.append(tag)
        # 创建条目并关联标签，ORM自动处理中间表插入
        entry = Entry(name=entry_name, tags=tag_objs)
        session.add(entry)
        session.commit()
        return entry.id

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