How to import a CSV file through a bulk ingest job from a script#

The HR system exports its contacts to a CSV every night, and your script uploads the file to the vendor’s bulk ingest job. Some mornings the job reads Failed, with a message about a field name you can see is spelled correctly. Other mornings it reads JobComplete with 11,940 of 12,000 rows in, so you open the vendor’s console to look for the other 60 by hand.

What you get

You will end up with a script that normalizes the export, drives the job to a terminal state, and writes a rejects CSV naming each missing row, its line, and its error. This is for you if a bulk job loads another system’s export.

Short answer

A Bulk API 2.0 ingest job takes four calls: create the job, PUT the CSV, PATCH the state to UploadComplete, and poll until JobComplete, Failed, or Aborted. Before the upload, decode the export with utf-8-sig and rewrite its line endings to the one the job declares. Afterwards, download the failed and unprocessed result files, match each row to its line by an external id, and refuse to report success unless rows in equals succeeded plus failed plus unprocessed.

You will need

Python 3.11 or later, an API that offers a bulk ingest job, and a CSV another system exported. Verified 2026-10-21 against Python 3.13.16. The sample uses the standard library alone: csv rewrites the file, urllib.request makes the calls, and a fake org built on http.server answers on 127.0.0.1, so no Salesforce account is involved. The calls and their order follow the Trailhead unit on Bulk API 2.0.

Approaches compared

The job trades time for accounting. A job answers nothing about a row until it ends; in return, its three result files name every row the file held. A synchronous batch endpoint such as Elasticsearch’s _bulk answers per chunk, so a bad row is known within a second of being sent. The price is a request per chunk and chunking code of your own. One request per row is the simplest to write and the most expensive to run, because the request count equals the row count.

Normalize the file before it leaves your machine

An export from a Windows tool carries two things a job declared LF cannot read. A UTF-8 byte order mark sits glued to the first header name, and CRLF line endings leave a carriage return on the last field of every row. A header name the object does not have fails the whole job rather than one row, because every row maps through that header.

def normalize_csv(raw: bytes, line_ending: str = "LF") -> tuple[bytes, list[dict[str, str]]]:
    """Drop a UTF-8 byte order mark, rewrite every line ending to the one the job will declare,
    and drop blank lines. Returns the bytes to upload and the rows, keyed by the header."""
    # utf-8-sig skips a BOM. Plain utf-8 keeps it, glued to the first header name.
    text = raw.decode("utf-8-sig")
    # newline="" hands the csv module the raw line endings, which it reads whether
    # they are CRLF, LF or a mixture of the two.
    rows = [r for r in csv.reader(io.StringIO(text, newline="")) if any(c.strip() for c in r)]
    if not rows:
        raise ValueError("the file has no header row")
    out = io.StringIO()
    csv.writer(out, lineterminator=LINE_ENDINGS[line_ending]).writerows(rows)
    header, body = rows[0], rows[1:]
    return out.getvalue().encode("utf-8"), [dict(zip(header, r)) for r in body]

utf-8-sig reads the mark as a signature and drops it,¹ where plain utf-8 keeps it as a character. Opening the text with newline="" hands the csv module the raw endings, which it reads whether they are CRLF, LF, or a mixture. The writer then emits the one ending the job will be told about. CRLF is the ending the CSV specification prescribes,² which is why so many exports carry it.

python3 demo.py

normalize contacts.csv
  in:  593 bytes, BOM yes, CRLF 13, bare LF 0
  out: 577 bytes, BOM no, CRLF 0, bare LF 13, rows 12

Sixteen bytes left the file: three for the mark and one carriage return for each of thirteen lines.

Create, upload, close, then poll

Four calls move the file into the job.

    def create_job(self, object_name: str, operation: str = "insert", line_ending: str = "LF",
                   external_id_field: str | None = None) -> dict:
        spec = {"object": object_name, "operation": operation, "contentType": "CSV",
                "lineEnding": line_ending}
        if external_id_field:
            spec["externalIdFieldName"] = external_id_field
        _, body = self._call("POST", f"{self.api}/jobs/ingest", json.dumps(spec).encode())
        return json.loads(body)

    def upload(self, job: dict, data: bytes) -> None:
        """PUT the CSV to the job's contentUrl as text/csv. The guide's example contentUrl
        carries no leading slash, so one is added."""
        status, _ = self._call("PUT", "/" + job["contentUrl"].lstrip("/"), data, "text/csv")
        if status != 201:
            raise BulkApiError("PUT", job["contentUrl"], status, "expected 201 Created")

    def set_state(self, job_id: str, state: str) -> dict:
        _, body = self._call("PATCH", f"{self.api}/jobs/ingest/{job_id}",
                             json.dumps({"state": state}).encode())
        return json.loads(body)

    def close(self, job_id: str) -> dict:
        return self.set_state(job_id, "UploadComplete")

    def abort(self, job_id: str) -> dict:
        return self.set_state(job_id, "Aborted")

The job is created with contentType set to CSV and the lineEnding the normalizer wrote. Its response carries a contentUrl, and the upload is a PUT of the bytes to that URL as text/csv, answered with 201 Created and no body.³ Processing starts only when the state is patched to UploadComplete. Aborting is the same PATCH with a different state.

def wait(client: BulkJobClient, job_id: str, *, max_polls: int = 120, interval: float = 5.0,
         sleep=time.sleep, abort_at_deadline: bool = False) -> tuple[dict, list[str]]:
    """Poll until a terminal state. At the deadline either abort the job, which turns the rows
    it never reached into a downloadable file, or raise with the job id so a caller can return."""
    states: list[str] = []
    for n in range(max_polls):
        job = client.info(job_id)
        states.append(job["state"])
        if job["state"] in TERMINAL:
            return job, states
        if n + 1 < max_polls:
            sleep(interval)
    if not abort_at_deadline:
        raise JobTimeout(job_id, states[-1])
    try:
        job = client.abort(job_id)
    except BulkApiError:
        # The job can end between the last poll and the abort, and an ended job refuses it.
        job = client.info(job_id)
        if job["state"] not in TERMINAL:
            raise
    states.append(job["state"])
    return job, states

The status endpoint sends no Retry-After, so the schedule is yours; the prerequisite page covers backoff, and this loop keeps a fixed interval with sleep injectable so the tests skip the wait. At the deadline there are two defensible moves. Raise with the job id, so the caller can come back for a job that is still running. Or abort it, so the rows the job never reached become the unprocessed file while the rows it did process stay written. The job can end between the last poll and the abort, and an ended job refuses it, so wait reads the job again before it gives up.

python3 demo.py

job 1: the import, a job that completes
  POST /jobs/ingest -> 200
  PUT /jobs/ingest/750fake00000000001/batches -> 201
  PATCH /jobs/ingest/750fake00000000001 -> 200
  GET /jobs/ingest/750fake00000000001 -> 200
  GET /jobs/ingest/750fake00000000001 -> 200
  GET /jobs/ingest/750fake00000000001 -> 200
  GET /jobs/ingest/750fake00000000001 -> 200
  GET /jobs/ingest/750fake00000000001/successfulResults/ -> 200
  GET /jobs/ingest/750fake00000000001/failedResults/ -> 200
  GET /jobs/ingest/750fake00000000001/unprocessedrecords/ -> 200
  states: UploadComplete, InProgress, InProgress, JobComplete
  rows in 12 = succeeded 10 + failed 2 + unprocessed 0

Ten requests for twelve rows: three to get the data in, four polls, and three downloads. Only the polls grow with the file, so seven requests is the least a job can cost.

Match every reject back to its line

The result files promise no order, and the fake returns them reversed to make the point. The match therefore runs on the external id column, which an upsert needs anyway. import_csv checks before the upload that the column is present, filled in, and unique, because a repeated value would point a reject at the wrong line.

def rejects_for(rows: list[dict], failed: list[dict], unprocessed: list[dict],
                key: str) -> list[dict[str, str]]:
    """The original row, the line it came from, and why it is not in the system. Rows are matched
    by the key column because the result files promise no order. Line 1 is the header."""
    header = list(rows[0].keys()) if rows else []
    line_of = {row[key]: n for n, row in enumerate(rows, start=2)}
    out = []
    for result, reason in [(r, r.get("sf__Error", "")) for r in failed] + \
                          [(r, UNPROCESSED) for r in unprocessed]:
        row = {name: result.get(name, "") for name in header}
        out.append({**row, "line": str(line_of.get(result.get(key), "")), "error": reason})
    out.sort(key=lambda r: int(r["line"] or 0))
    return out

The failed file carries the vendor’s sf__Error beside the original columns, and the unprocessed file carries the original columns alone, so the second group gets a reason of its own. The check that closes the import is one equation.

    ok = client.results(job["id"], "successfulResults")
    failed = client.results(job["id"], "failedResults")
    unprocessed = client.results(job["id"], "unprocessedrecords")
    outcome = Outcome(final, states, list(rows[0].keys()) if rows else [], len(rows),
                      len(ok), len(failed), len(unprocessed),
                      rejects_for(rows, failed, unprocessed, key))
    missing = outcome.rows_in - outcome.succeeded - outcome.failed - outcome.unprocessed
    if missing:
        raise AccountingError(f"{outcome.summary()}: {missing} rows have no result")
    return outcome

A row that is in none of the three files has no result at all, and a result file is a download like any other, so it can come back short. AccountingError stops the script before it reports a success it cannot show.

python3 demo.py

rejects.csv
External_Id__c,FirstName,LastName,Email,line,error
HR-1004,Maria,Gonzalez,maria.gonzalez.example.com,5,INVALID_EMAIL_ADDRESS:Email: invalid email address: maria.gonzalez.example.com:Email --
HR-1009,Joseph,Okafor,j.okafor_at_example.com,10,INVALID_EMAIL_ADDRESS:Email: invalid email address: j.okafor_at_example.com:Email --

job 2: the same file, a deadline of two polls, then abort
  states: UploadComplete, InProgress, Aborted
  rows in 12 = succeeded 4 + failed 1 + unprocessed 7
  rejects: 1 failed, 7 unprocessed, lines 5, 7, 8, 9, 10, 11, 12, 13

Each reject is the original row, its line in the uploaded file, and the error text. The second job is the same file with the deadline set to two polls. The abort leaves seven rows unprocessed and the equation still closes. The rows the job reached are written, the rest sit in the unprocessed file, and the rejects name both kinds with a reason you can filter on.

Measure when you learn about a bad row

compare.py pushes the same twelve rows through the job, through a synchronous collections endpoint five rows a call, and one request at a time with three in flight.

python3 compare.py

12 rows, 2 with a bad email, on lines 5 and 10 of the file

approach                             requests  accepted  rejected  when you learn about line 5
bulk ingest job                            10        10         2  after JobComplete, in failedResults
synchronous batch, 5 rows a call            3        10         2  in the response to call 1
one request per row, 3 in flight           12        10         2  in the response to its own request

Read the last column. The job reports line 5 only after it has ended and a result file has been fetched: the tenth request here, and minutes later in production. The batch endpoint reports it in its first response, and the per-row approach in the response to line 5 itself, at a request per row.

Check it worked

Run the cases. The sixth one matters most: it points the importer at a fake whose failed-results file is one row short, and asserts that import_csv raises rather than reports. The last three refuse a repeated key before the upload, survive an abort that arrives after the job ended, and time out on a server that never answers.

python3 test_import.py

ok a_bom_and_crlf_endings_are_normalized_away
ok every_row_is_accounted_for_after_jobcomplete
ok a_reject_carries_the_original_row_its_line_and_the_error
ok an_aborted_job_leaves_unprocessed_rows_that_still_sum
ok a_header_the_object_does_not_have_fails_the_whole_job
ok a_short_result_file_raises_instead_of_reporting_success
ok a_repeated_or_missing_key_is_refused_before_anything_is_sent
ok an_abort_that_loses_the_race_reads_the_job_again
ok a_server_that_never_answers_times_out
9 cases, 0 failures

9 cases, 0 failures is the line to look for. The fourth case pins the aborted job to 4 + 1 + 7 and the lines of its unprocessed rows to 7 through 13, so a matching bug that lost a row would fail there.

When it goes wrong

The job reads Failed before it processed a row. Upload the export as it came, and the header is wrong in two places at once.

python3 pitfall.py

the file as exported, job declared LF
  header as the server read it: ['\ufeffExternal_Id__c', 'FirstName', 'LastName', 'Email\r']
  states: UploadComplete, Failed
  errorMessage: 'InvalidBatch : Field name not found : \ufeffExternal_Id__c'
  succeeded 0, failed 0, unprocessed 12

The first header name has \ufeff in front of it, the invisible mark, and the last has \r behind it, the half of CRLF that a split on LF leaves in place. Neither is a field the object has, so the job fails at the header and all twelve rows sit in the unprocessed file. The error names a field that looks right in every editor.

The loop never ends. A job can sit in UploadComplete for longer than you planned, and the status endpoint sends no Retry-After to pace you. Set max_polls, and decide in advance whether the deadline aborts the job or raises with its id.

The counts do not sum. AccountingError says how many rows have no result, which means a result file came back short. Download the three files again by the job id before you treat any row as lost.

When not to do this

Do not put a few hundred rows through a job. The price is at least seven requests and a wait, and a synchronous batch endpoint answers the same rows in one call with the errors in the response.

Do not abort at the deadline against a vendor whose job offers no unprocessed file. The abort is safe here because unprocessedrecords names what was never attempted; without it, raising with the job id is the only move left.

Do not rerun a whole file after Failed or Aborted with an insert operation. The rows processed before the stop stay written, as job 2 shows, so a rerun duplicates them. Use upsert on the external id, or rerun the rejects file alone.

Do not filter rejects on the error sentence. The sentence belongs to the vendor, who may reword it. Filter on the code before the first colon, which is the part written to be matched.

Related how-tos

• Upload a large file with resumable chunks

• Report partial failure in a batch endpoint

• Poll a job status endpoint with backoff

• Batch writes without losing per-item error detail

Last verified

Verified 2026-10-21 against Python 3.13.16. Every output block is what the command preceding it printed, against the fake org in the code directory and not against a Salesforce org. The paths, state names, result files, and sf__ columns follow the Bulk API 2.0 developer guide. The error texts, the header check, the reversed result order, and the pace of processing are the fake’s own.