Supabase Python Upsert多列唯一约束冲突处理求助

你的核心问题是Supabase的upsert API不支持同时指定多个独立的唯一约束作为冲突判断条件。PostgreSQL本身的ON CONFLICT语法只能针对单个冲突目标（单个唯一列、复合唯一约束或约束名称），而Supabase的API直接映射了这个限制：

• 第一次尝试的doi,(title,venue),(title,abstract)格式不符合API解析规则，触发PGRST100解析错误；
• 第二次尝试的doi,title,venue,abstract没有对应的唯一约束（表中只有三个独立的唯一约束：doi、(title,venue)、(title,abstract)，没有包含这四个列的联合约束），触发42P10约束不匹配错误。

要处理多个独立唯一约束的冲突更新，需要绕过Supabase的基础upsert API，改用自定义PostgreSQL函数+Supabase RPC调用，或者直接执行原生SQL语句。

方案1：创建自定义PL/pgSQL函数处理单条数据 #

先在Supabase的SQL编辑器中创建函数，逐个检查所有冲突条件并执行对应更新/插入操作：

CREATE OR REPLACE FUNCTION upsert_paper(p_title text, p_abstract text, p_venue text, p_doi text)
RETURNS uuid AS $$
DECLARE
  existing_id uuid;
BEGIN
  -- 检查DOI冲突
  SELECT id INTO existing_id FROM public.paper WHERE doi = p_doi;
  IF existing_id IS NOT NULL THEN
    UPDATE public.paper
    SET title = p_title, abstract = p_abstract, venue = p_venue
    WHERE id = existing_id;
    RETURN existing_id;
  END IF;

  -- 检查标题+会议/期刊冲突
  SELECT id INTO existing_id FROM public.paper WHERE title = p_title AND venue = p_venue;
  IF existing_id IS NOT NULL THEN
    UPDATE public.paper
    SET abstract = p_abstract, doi = p_doi
    WHERE id = existing_id;
    RETURN existing_id;
  END IF;

  -- 检查标题+摘要冲突
  SELECT id INTO existing_id FROM public.paper WHERE title = p_title AND abstract = p_abstract;
  IF existing_id IS NOT NULL THEN
    UPDATE public.paper
    SET venue = p_venue, doi = p_doi
    WHERE id = existing_id;
    RETURN existing_id;
  END IF;

  -- 无冲突，插入新记录
  INSERT INTO public.paper(title, abstract, venue, doi)
  VALUES(p_title, p_abstract, p_venue, p_doi)
  RETURNING id INTO existing_id;
  RETURN existing_id;
END;
$$ LANGUAGE plpgsql;

然后在Python中通过Supabase的rpc方法调用该函数，处理每条数据（冲突时更新，无冲突时插入，错误时跳过）：

paper_id_map = {}
for paper in papers_without_authors:
    try:
        res = supabase.rpc(
            "upsert_paper",
            {
                "p_title": paper["title"],
                "p_abstract": paper.get("abstract"),
                "p_venue": paper.get("venue"),
                "p_doi": paper.get("doi")
            }
        ).execute()
        paper_id_map[paper["title"]] = res.data
    except Exception as e:
        print(f"跳过论文 {paper['title']}: {str(e)}")
        continue

方案2：批量处理的原生SQL（适合大数据量） #

如果爬虫获取的数据量很大，循环调用函数效率较低，可以用UNNEST批量传递参数，结合函数实现批量处理：

1. 创建批量处理函数：

CREATE OR REPLACE FUNCTION batch_upsert_papers(
  p_titles text[],
  p_abstracts text[],
  p_venues text[],
  p_dois text[]
) RETURNS uuid[] AS $$
DECLARE
  result_ids uuid[];
  i integer;
  current_id uuid;
BEGIN
  result_ids := '{}'::uuid[];
  FOR i IN 1..array_length(p_titles, 1) LOOP
    current_id := upsert_paper(
      p_titles[i],
      p_abstracts[i],
      p_venues[i],
      p_dois[i]
    );
    result_ids := array_append(result_ids, current_id);
  END LOOP;
  RETURN result_ids;
END;
$$ LANGUAGE plpgsql;

2. 在Python中批量组织参数并调用：

# 整理批量参数
titles = [p["title"] for p in papers_without_authors]
abstracts = [p.get("abstract") for p in papers_without_authors]
venues = [p.get("venue") for p in papers_without_authors]
dois = [p.get("doi") for p in papers_without_authors]

try:
    res = supabase.rpc(
        "batch_upsert_papers",
        {
            "p_titles": titles,
            "p_abstracts": abstracts,
            "p_venues": venues,
            "p_dois": dois
        }
    ).execute()
    # 映射标题到ID
    paper_id_map = {titles[i]: res.data[i] for i in range(len(titles))}
except Exception as e:
    print(f"批量处理失败: {str(e)}")
    # 可选择降级为单条处理

• 函数中的更新逻辑可以根据需求调整（比如只更新非空字段）；
• 若爬虫是多线程/多进程运行，建议在函数中添加事务控制，避免并发冲突；
• 若要跳过冲突行而非更新，只需将函数中的UPDATE语句替换为直接返回existing_id即可。

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