require "csv"
class ProductImporter
BATCH_SIZE = 500
attr_reader :errors, :inserted
def initialize(io)
@io = io
@buffer = []
@errors = []
@inserted = 0
end
def call
CSV.new(@io, headers: true).each.with_index(2) do |row, line|
attrs = build_attributes(row)
@buffer << attrs if attrs
rescue KeyError, ArgumentError => e
@errors << { line: line, message: e.message }
ensure
flush_batch if @buffer.size >= BATCH_SIZE
end
flush_batch
self
end
private
def build_attributes(row)
sku = row["sku"].to_s.strip.presence
raise ArgumentError, "missing sku" unless sku
{
sku: sku,
name: row["name"].to_s.strip,
price_cents: Integer(Float(row["price"].to_s) * 100),
currency: row["currency"].presence || "USD",
updated_at: Time.current
}
end
def flush_batch
return if @buffer.empty?
result = Product.upsert_all(
@buffer,
unique_by: :sku,
update_only: %i[name price_cents currency updated_at]
)
@inserted += result.rows.size
@buffer.clear
end
end
class ProductImportJob < ApplicationJob
queue_as :imports
retry_on ActiveRecord::StatementInvalid, wait: :polynomially_longer, attempts: 3
def perform(import_record_id)
record = ImportRecord.find(import_record_id)
record.update!(status: :running, started_at: Time.current)
importer = record.file.open do |file|
ActiveRecord::Base.transaction do
ProductImporter.new(file).call
end
end
record.update!(
status: importer.errors.any? ? :completed_with_errors : :completed,
inserted_count: importer.inserted,
error_report: importer.errors,
finished_at: Time.current
)
rescue => e
record&.update!(status: :failed, error_report: [{ message: e.message }])
raise
end
end
class ImportsController < ApplicationController
def create
upload = params.require(:file)
record = ImportRecord.create!(
kind: :products,
status: :pending,
original_filename: upload.original_filename
)
record.file.attach(upload)
ProductImportJob.perform_later(record.id)
render json: { id: record.id, status: record.status }, status: :accepted
end
def show
record = ImportRecord.find(params[:id])
render json: record.slice(:id, :status, :inserted_count, :error_report)
end
end
This snippet shows how a CSV product import is structured in Rails so that thousands of rows land in the database in a handful of round trips instead of one INSERT per row. The work is split across a service object that does the heavy lifting, a background job that runs it off the request cycle, and a thin controller that accepts the upload and hands it off.
In ProductImporter service, the CSV is parsed with the standard library CSV class using headers: true so each row behaves like a hash keyed by column name. Rows are accumulated into an in-memory buffer and flushed in fixed-size batches via flush_batch. The flush uses ActiveRecord's upsert_all, which compiles a single multi-row INSERT ... ON CONFLICT statement. This is what makes the import fast: it bypasses model instantiation and validations, so callbacks never fire and each batch is one SQL statement. The trade-off is deliberate — because validations are skipped, the service normalizes and guards data itself in build_attributes, coercing the price and rejecting rows without a sku.
Idempotency comes from unique_by: :sku combined with on_duplicate update logic. Re-running the same file updates existing products by their natural key rather than creating duplicates, which matters because imports get retried. The updated_at timestamp is set explicitly since upsert_all does not touch timestamps automatically. Errors are collected per-row into @errors instead of aborting the whole run, so one malformed line does not discard an otherwise good file.
In ProductImportJob, the service is invoked inside a transaction and the resulting summary is persisted back to an ImportRecord, giving the user a durable status and error report to poll. Wrapping the batches in one transaction means a mid-import failure rolls back cleanly. ActiveJob retries handle transient database hiccups.
In ImportsController, the uploaded file is streamed to storage and only its identifier is passed to the job — never the raw file through the queue, which keeps job payloads small and serializable. The controller responds with 202 Accepted, the correct status for work that completes asynchronously. Together these files show the common Rails pattern of a fast, validation-light bulk path fronted by an explicit service and kept off the web thread.
Related snips
class CommentsController < ApplicationController
before_action :set_post
def create
@comment = @post.comments.build(comment_params)System test: asserting Turbo Stream responses
class Post < ApplicationRecord
belongs_to :author, class_name: 'User'
has_many :comments, dependent: :destroy
scope :published, -> { where.not(published_at: nil).where('published_at <= ?', Time.current) }
scope :draft, -> { where(published_at: nil) }ActiveRecord scopes for reusable query logic
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class PostsController < ApplicationController
def index
@posts = Post.includes(:author)
.order(created_at: :desc)
.page(params[:page])
.per(10)Turbo Frames: infinite scroll with lazy-loading frame
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
Share this code
Here's the card — post it anywhere.
