Importação massiva de CSV com Npgsql COPY em Minimal APIs e EF Core

1) Motivacao e trade-offs

Por que usar COPY em vez de INSERT em lote?

O comando COPY do PostgreSQL é otimizado para transferências em massa. Para cargas grandes — dezenas de milhares a milhões de linhas — COPY costuma ser dezenas de vezes mais rápido que executar INSERTs em lote via EF Core ou ADO.NET. Ele minimiza parsing SQL, reduz o número de round-trips e aproveita caminhos internos do servidor.

Mesmo assim, COPY tem trade-offs: é mais rígido em relação ao formato (CSV, texto, binary), menos integrado com o modelo de entidades do EF Core e exige cuidado com transações, tipos e validações. Use COPY quando desempenho e throughput forem críticos; caso contrário, INSERTs em lote ou ORM podem ser mais simples.

Trade-offs operacionais

Escopos principais a considerar:

• Validação: VALIDATE antes do COPY (client-side) ou após o COPY em uma etapa separada.
• Atomicidade: você pode rodar o COPY dentro de uma transação para garantir rollback se algo falhar.
• Mapeamento: o COPY escreve diretamente para tabelas; não respeita ganchos EF (ChangeTracker, Events).
• Segurança: evite COPY FROM FILE no servidor, prefira STDIN com streaming controlado pelo cliente.

Use COPY para throughput. Use EF Core para regras de domínio que exigem lógica por entidade. Combine: COPY para bulk ingest e depois um processo de normalização/merge via EF/SQL.

2) Fundamentos tecnicos

Como o PostgreSQL COPY funciona

COPY importa/exports dados entre uma tabela e um arquivo ou STDIN/STDOUT. Suporta formatos CSV e BINARY. Quando usado via STDIN, o cliente abre um stream e envia os bytes; o servidor processa e popula a tabela. Npgsql expõe helpers para trabalhar com COPY de forma eficiente, sem construir SQL de INSERT em massa.

No modo CSV, o alinhamento das colunas e a representação de NULL, delimitador e escape devem ser exatamente os esperados pelo servidor. Em BINARY, há ganho de performance maior e segurança contra parsing de CSV, porém exige conversão de tipos binários.

Integração com Npgsql e EF Core

Npgsql oferece APIs para iniciar operações COPY (text/binary). Em um fluxo típico com Minimal APIs você receberá o arquivo HTTP, fará Streaming para o COPY e, em seguida, usará EF Core para aplicar merges/constraints ou para atualizar índices e caches.

Pattern recomendado: STREAM -> COPY para tabela temporária -> validação/merge dentro da mesma transação -> commit. Isso minimiza janela inconsistente e permite rollback completo em caso de erro.

3) Exemplo 1: caso simples

Objetivo do exemplo

Endpoint minimal API que recebe um arquivo CSV via form-data e usa COPY para inserir diretamente em uma tabela simples 'products'. Foco: simplicidade e correção para cargas moderadas.

Estrutura da tabela

Tabela PostgreSQL utilizada no exemplo:

-- SQL no servidor (execute via psql ou migraçao)
CREATE TABLE public.products (
  sku text PRIMARY KEY,
  name text NOT NULL,
  price numeric(10,2) NOT NULL,
  created_at timestamptz DEFAULT now()
);

Agora o endpoint Minimal API (arquivo Program.cs). O bloco abaixo recebe o arquivo e faz COPY via stream. Comentários explicam o que acontece antes e depois.

// O que faz: recebe IFormFile CSV e copia para a tabela products usando COPY ... FROM STDIN (CSV)
// Após o COPY, retorna contagem de linhas importadas (estimada pelo server).
app.MapPost("/import/products", async (HttpRequest request, IConfiguration config, CancellationToken ct) => {
    if (!request.HasFormContentType) return Results.BadRequest("form-data expected");
    var form = await request.ReadFormAsync(ct);
    var file = form.Files.GetFile("file");
    if (file == null) return Results.BadRequest("file= is required");

    var connString = config.GetConnectionString("Default");
    await using var conn = new Npgsql.NpgsqlConnection(connString);
    await conn.OpenAsync(ct);

    // Monta o comando COPY. Se o CSV contém header, devemos pular a primeira linha
    var copySql = "COPY public.products (sku, name, price) FROM STDIN (FORMAT csv, DELIMITER ',', NULL '')";
    try
    {
        await using var importer = conn.BeginTextImport(copySql);
        // Ler o arquivo e enviar linhas (pulando header se presente)
        await using var sr = new StreamReader(file.OpenReadStream());
        var firstLine = await sr.ReadLineAsync();
        var hasHeader = firstLine != null && firstLine.Contains("sku");
        if (!hasHeader && firstLine != null)
            await importer.WriteLineAsync(firstLine);

        string? line;
        while ((line = await sr.ReadLineAsync()) != null)
        {
            await importer.WriteLineAsync(line);
        }

        // Depois do using o COPY é finalizado e o server processa os dados.
        return Results.Ok(new { message = "import completed" });
    }
    catch (Npgsql.NpgsqlException ex)
    {
        // O servidor envia mensagens úteis em ex.Message
        return Results.StatusCode(500, new { error = ex.Message });
    }
});

O que acontece após a execução: o COPY insere as linhas na tabela. Se ocorrer um erro sintático ou de tipo na linha X, o servidor aborta e a exception traz a linha e o motivo na mensagem.

4) Exemplo 2: caso avancado do mundo real

Objetivo do caso real

Processo de ingestão para sincronizar dados de um parceiro. Requisitos: 1) alta taxa de ingestão; 2) validação e enriquecimento (normalização SKU, cálculo de margens); 3) evitar downtime da tabela principal; 4) aplicar upsert (insert ou update) por chave natural.

Estratégia adotada

Fluxo:

1. Receber CSV e COPY para uma tabela temporária (temp table ou schema staging).
2. Executar validações e transformações com SQL ou C# (EF Core raw SQL para operações em lote).
3. Executar MERGE/UPSERT com ON CONFLICT para aplicar para a tabela principal.
4. Manter tudo em uma transação para garantir atomicidade.

Vantagens: validação sem poluir a tabela principal; reversão simples; permite índices temporários para acelerar merge se necessário.

A seguir um exemplo completo de endpoint que implementa o fluxo. Assume EF Core DbContext chamado AppDbContext.

// O que faz: usa COPY para inserir em um schema staging.products_staging,
// valida e faz upsert na tabela public.products usando ON CONFLICT.
app.MapPost("/import/products/advanced", async (HttpRequest request, AppDbContext db, IConfiguration config, CancellationToken ct) => {
    var form = await request.ReadFormAsync(ct);
    var file = form.Files.GetFile("file");
    if (file == null) return Results.BadRequest("file required");

    var connString = config.GetConnectionString("Default");
    await using var conn = new Npgsql.NpgsqlConnection(connString);
    await conn.OpenAsync(ct);

    // Usamos transação para garantir rollback se algo falhar
    await using var tx = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted, ct);
    try
    {
        // 1) Criar tabela de staging temporária (schema staging) - pode ser schema permanente e truncado
        var createSql = @"
            CREATE TEMP TABLE products_staging (
                sku text,
                name text,
                price numeric(10,2)
            ) ON COMMIT DROP
        ";
        await using (var cmd = new Npgsql.NpgsqlCommand(createSql, conn, tx))
            await cmd.ExecuteNonQueryAsync(ct);

        // 2) COPY para staging
        var copySql = "COPY products_staging (sku, name, price) FROM STDIN (FORMAT csv, DELIMITER ',', NULL '')";
        await using (var importer = conn.BeginTextImport(copySql))
        await using (var sr = new StreamReader(file.OpenReadStream()))
        {
            // pular header inteligente
            var header = await sr.ReadLineAsync();
            var hasHeader = header != null && header.Contains("sku");
            if (!hasHeader && header != null) await importer.WriteLineAsync(header);
            string? line;
            while ((line = await sr.ReadLineAsync()) != null)
            {
                // validação simples: skip linhas vazias
                if (string.IsNullOrWhiteSpace(line)) continue;
                await importer.WriteLineAsync(line);
            }
        }

        // 3) Validações e transformações em SQL para performance
        var validateSql = @"
            -- Exemplo: remover SKUs nulos e normalizar trim
            DELETE FROM products_staging WHERE sku IS NULL OR trim(sku) = '';
            UPDATE products_staging SET sku = trim(sku), name = trim(name);
        ";
        await using (var cmd = new Npgsql.NpgsqlCommand(validateSql, conn, tx))
            await cmd.ExecuteNonQueryAsync(ct);

        // 4) Upsert para tabela principal com ON CONFLICT
        var upsertSql = @"
            INSERT INTO public.products (sku, name, price)
            SELECT sku, name, price FROM products_staging
            ON CONFLICT (sku)
            DO UPDATE SET name = EXCLUDED.name, price = EXCLUDED.price;
        ";
        await using (var cmd = new Npgsql.NpgsqlCommand(upsertSql, conn, tx))
            var rows = await cmd.ExecuteNonQueryAsync(ct);

        await tx.CommitAsync(ct);
        return Results.Ok(new { imported = rows });
    }
    catch (Exception ex)
    {
        await tx.RollbackAsync(ct);
        return Results.Problem(detail: ex.Message);
    }
});

O que acontece: a operação é feita integralmente no servidor dentro de uma transação; a tabela temporária é descartada ao final. A cláusula ON CONFLICT aplica o upsert. Você obtém alta taxa de ingestão pelo COPY e consistência pelo uso da transação.

5) Debugging e troubleshooting

Erros comuns e armadilhas

Listagem de erros que aparecem com frequência ao usar COPY:

• Formato CSV inválido: escape ou delimitador inconsistente gera erro de parsing e falha no COPY.
• Types mismatch: enviar string em coluna numeric sem conversão ou com vírgula como separador decimal.
• Header presente quando o COPY não espera header (ou vice-versa).
• Conflitos de integridade referencial ou chave primária ao inserir dados duplicados.
• Problemas de transação: COPY fora de uma transação pode deixar dados parciais se falhar em etapas posteriores.

Sempre logue a mensagem completa da NpgsqlException. Ela traz a linha e o detalhe do erro do PostgreSQL, crucial para localizar problemas de parsing ou tipo.

Técnicas de investigação

1) Reproduza localmente com um arquivo pequeno e psql: execute COPY ... FROM STDIN e cole um trecho do CSV para ver o erro do servidor. Isso isola se o problema é de rede/streaming ou do conteúdo.

2) Teste schema em staging: copie para uma tabela temporária e execute queries de validação para detectar linhas problemáticas com LIMIT e OFFSET.

3) Valide tipos: crie um SELECT que tenta cast nas colunas para expor erros: SELECT (col::numeric) FROM staging LIMIT 1000.

Exemplo de captura de erro e extração de context do servidor:

// O que faz: captura NpgsqlException e extrai detalhes úteis para log
try
{
    // executar COPY
}
catch (Npgsql.NpgsqlException nex)
{
    // nex.Message traz detalhe do PostgreSQL. nex.ErrorCode / nex.SqlState podem ajudar.
    logger.LogError(nex, "Copy failed: sqlstate={SqlState}", nex.SqlState);
    throw; // rethrow ou adaptar resposta ao cliente
}

Conclusao

Resumo prático

COPY via Npgsql é a escolha natural para importações massivas em ASP.NET Core Minimal APIs quando a prioridade é throughput. Combine COPY para ingestão bruta com SQL/EF Core para validação e upsert. Execute COPY dentro de transações quando precisar de atomicidade e prefira tabelas de staging para minimizar impacto na tabela principal.

Checklist antes de produção

• Confirme o encoding do CSV (UTF-8 preferível).
• Padronize delimitadores e representação de NULL.
• Implemente logging detalhado para NpgsqlException.
• Teste com amostras grandes localmente para aferir performance.
• Considere COPY BINARY se precisar de performance máxima e tiver controle de tipo.

Implementar COPY corretamente reduz custo CPU no cliente e melhora latência por linha. Invista um pouco na etapa de staging/validação e você terá um pipeline robusto e escalável para importação de CSV em produção.

Nota final: o código mostrado usa APIs comuns do Npgsql e ADO.NET dentro do paradigma Minimal APIs. Adapte a estratégia para volumes, regras de negócio e requisitos de consistência do seu sistema.