Rivulet.Sql.PostgreSql¶

using Npgsql;
using Rivulet.Sql.PostgreSql;

var users = GetUsers(); // IEnumerable<User>

await users.BulkInsertUsingCopyAsync(
    () => new NpgsqlConnection(connectionString),
    "users",
    columns: new[] { "id", "name", "email" },
    mapToRow: user => new object?[] { user.Id, user.Name, user.Email },
    options: new ParallelOptionsRivulet { MaxDegreeOfParallelism = 4 },
    batchSize: 5000
);

var csvLines = File.ReadLines("users.csv");

await csvLines.BulkInsertUsingCopyCsvAsync(
    () => new NpgsqlConnection(connectionString),
    "users",
    columns: new[] { "id", "name", "email" },
    options: new ParallelOptionsRivulet { MaxDegreeOfParallelism = 2 },
    hasHeader: true,
    delimiter: ','
);

var textLines = File.ReadLines("users.txt");

await textLines.BulkInsertUsingCopyTextAsync(
    () => new NpgsqlConnection(connectionString),
    "users",
    columns: new[] { "id", "name", "email" },
    options: new ParallelOptionsRivulet { MaxDegreeOfParallelism = 4 },
    batchSize: 10000
);

record Product(int Id, string Name, decimal Price, DateTime CreatedAt);

var products = GetProducts();

await products.BulkInsertUsingCopyAsync(
    () => new NpgsqlConnection(connectionString),
    "products",
    columns: new[] { "id", "name", "price", "created_at" },
    mapToRow: product => new object?[]
    {
        product.Id,
        product.Name,
        product.Price,
        product.CreatedAt
    },
    options: new ParallelOptionsRivulet
    {
        MaxDegreeOfParallelism = 4,
        ErrorMode = ErrorMode.CollectAndContinue
    }
);