如何在Supabase边缘函数中通过COPY命令从存储的CSV文件批量导入5万条数据？

阿华AIGC实验室

2026-4-16

嘿，我刚好折腾过类似的批量导入需求，你的整体流程思路是对的，但这里有个关键坑得先给你指出来：PostgreSQL的COPY命令没法直接从HTTP URL读取文件，你原来想直接用Signed URL的写法是走不通的，得换个方式把CSV内容先拉到边缘函数里，再传给数据库。

我给你梳理下修正后的完整实现步骤和代码细节：

核心问题修正

你之前的思路里，想用COPY reports FROM 'signedUrl'，但PostgreSQL的原生COPY只支持本地文件、标准输入（STDIN）或者程序输出，不支持直接拉取HTTP资源，所以必须先把CSV文件下载到边缘函数的内存中，再通过STDIN传给COPY命令。

完整实现步骤

1. 边缘函数里生成Signed URL并下载CSV

首先用Supabase Storage的Admin SDK生成带有效期的Signed URL（建议设5分钟，足够处理4MB的文件），然后用fetch把CSV内容拉到边缘函数里：

import { createSupabaseClient } from 'https://esm.sh/@supabase/supabase-js@2';
import { Client } from 'https://deno.land/x/postgres/mod.ts';

export default async function handler(req: Request) {
  // 1. 解析触发事件（从job插入的webhook拿到文件路径和job ID）
  const { record } = await req.json();
  const { file_path, id: jobId } = record;

  // 2. 初始化Supabase Admin客户端和Storage
  const supabase = createSupabaseClient(
    Deno.env.get('SUPABASE_URL')!,
    Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
  );
  // 生成5分钟有效期的Signed URL
  const { data: signedUrlData, error: urlError } = await supabase.storage
    .from('your-csv-bucket')
    .createSignedUrl(file_path, 300);
  
  if (urlError) {
    // 更新job状态为失败
    await supabase.from('jobs').update({ status: 'failed', error: urlError.message }).eq('id', jobId);
    return new Response(JSON.stringify({ error: urlError.message }), { status: 500 });
  }

  // 3. 下载CSV文件到边缘函数内存
  const csvRes = await fetch(signedUrlData.signedUrl);
  if (!csvRes.ok) {
    await supabase.from('jobs').update({ status: 'failed', error: '下载CSV失败' }).eq('id', jobId);
    return new Response('下载CSV失败', { status: 500 });
  }
  const csvContent = await csvRes.text();

2. 用PostgreSQL客户端执行COPY命令

接下来用deno的postgres库连接数据库，通过STDIN传入CSV内容完成批量导入：

// 4. 初始化PostgreSQL Admin客户端
  const client = new Client({
    connectionString: Deno.env.get('SUPABASE_DB_URL')!,
  });
  await client.connect();

  try {
    // 执行COPY命令，注意要和你的表字段、CSV格式匹配
    const copyResult = await client.copy(
      `COPY reports (column1, column2, column3) FROM STDIN WITH (
        FORMAT csv,
        HEADER true,  // 如果CSV第一行是表头就加这个，没有就删掉
        DELIMITER ',',
        ENCODING 'UTF8'
      )`,
      csvContent
    );

    // 5. 更新job状态为成功
    await supabase.from('jobs').update({
      status: 'completed',
      imported_count: copyResult.rowCount
    }).eq('id', jobId);

    return new Response(JSON.stringify({ success: true, rowCount: copyResult.rowCount }), { status: 200 });
  } catch (dbError) {
    await supabase.from('jobs').update({
      status: 'failed',
      error: dbError.message
    }).eq('id', jobId);
    return new Response(JSON.stringify({ error: dbError.message }), { status: 500 });
  } finally {
    // 不管成功失败都要关闭数据库连接
    await client.end();
  }
}