# How to import a CSV file through a bulk ingest job from a script > Normalize an exported CSV, run it through a Bulk API 2.0 ingest job, then download the failed and unprocessed rows and prove every row has exactly one result. Source: https://voxgig.com/howto/import-a-csv-through-a-bulk-job-api - Audience: api-consumer - Level: intermediate - Languages: python - Verified: 2026-10-21 - Published: 2026-10-17 ## 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`](https://docs.python.org/3/library/csv.html) rewrites the file, [`urllib.request`](https://docs.python.org/3/library/urllib.request.html) makes the calls, and a fake org built on [`http.server`](https://docs.python.org/3/library/http.server.html) 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](https://trailhead.salesforce.com/content/learn/modules/api_basics/api_basics_bulk). ## Approaches compared | Approach | When it fits | What it costs you | When to pick something else | | --- | --- | --- | --- | | [A Salesforce Bulk API 2.0 ingest job](https://trailhead.salesforce.com/content/learn/modules/api_basics/api_basics_bulk) | Thousands to millions of rows, where you can wait minutes and want three result files rather than a response per row | Three calls before processing starts, a poll loop the server gives no schedule for, and the first bad row reported only after the job ends | A few hundred rows, or a caller that needs the answer before it can continue | | [Chunked calls to a synchronous batch endpoint](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk) | Hundreds to tens of thousands of rows, where each response names the rows in that chunk that failed | A request per chunk against the rate limit, chunking and retry code that is yours to maintain, and a per-call all-or-nothing flag to get right | A file so large that the chunk loop runs for hours, which is the case the job exists to absorb | | [One request per row with bounded concurrency](https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor) | A few hundred rows, or rows whose outcome decides whether the next one is sent | A request per row against the rate limit, and a pool size tuned by hand against 429 responses | Anything beyond a few thousand rows, where the request count alone settles it | 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. ```py title="importer.py" 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,[^1] 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,[^2] which is why so many exports carry it. ```bash python3 demo.py ``` ```text output 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. ```py title="importer.py" 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.[^3] Processing starts only when the state is patched to `UploadComplete`. Aborting is the same `PATCH` with a different state. ```py title="importer.py" 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. ```bash python3 demo.py ``` ```text output 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. ```py title="importer.py" 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. ```py title="importer.py" 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. ```bash python3 demo.py ``` ```text output 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. ```bash python3 compare.py ``` ```text output 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. ```bash python3 test_import.py ``` ```text output 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. ```bash python3 pitfall.py ``` ```text output 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](/howto/upload-a-large-file-with-resumable-chunks) - [Report partial failure in a batch endpoint](/howto/report-partial-failure-in-a-batch-endpoint) - [Poll a job status endpoint with backoff](/howto/poll-a-job-status-endpoint-with-backoff) - [Batch writes without losing per-item error detail](/howto/batch-writes-without-losing-per-item-errors) ## 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. [^1]: Python's `codecs` documentation credits the variant to a text editor. [Microsoft invented a variant of UTF-8](https://docs.python.org/3/library/codecs.html#encodings-and-unicode), it says, for its Notepad program, writing the bytes `EF BB BF` before any character so a UTF-8 file could be told from a legacy encoding. The `sig` is short for signature, because the mark has nothing to say about byte order in an encoding with one byte per unit, a function [RFC 3629](https://www.rfc-editor.org/rfc/rfc3629.html#section-6) calls useless for UTF-8. The same section of the Python page closes by saying the mark is discouraged in UTF-8, two sentences after explaining how to write one. [^2]: [RFC 4180](https://www.rfc-editor.org/rfc/rfc4180.html), from October 2005 and Informational rather than a standard, registers the `text/csv` media type and describes what its title calls the common format. Its grammar ends every record with CRLF, and its second rule allows the last record to go without a line break at all. [^3]: [RFC 9110](https://www.rfc-editor.org/rfc/rfc9110.html#name-201-created) says a 201 means one or more new resources were created. The primary one is identified by a `Location` header or, when none is sent, by the target URI. The Trailhead unit describes the answer to the upload as [just a status code of 201 Created](https://trailhead.salesforce.com/content/learn/modules/api_basics/api_basics_bulk), so under that rule the resource created is the `batches` URL the CSV was put to.