“Bulk operation” is not one operation. Choose based on what must change and how each row differs. If every row is inserted, use a bulk insert. If rows may already exist and should be replaced or merged, use an upsert. If existing rows receive different values, use a bulk update strategy. If the same condition applies to every row, one set-based UPDATE is usually simpler and faster than sending one update per record.
For a modest collection of new rows, a parameterized executemany keeps the SQL shape stable and makes the intent plain. The driver may send these as separate statements or optimize them internally, so check its behavior rather than assuming it emits one network request.
rows = [
("a17", "Ada", True),
("b42", "Linus", True),
("c09", "Grace", False),
]
cursor.executemany(
"""
INSERT INTO users (user_id, display_name, active)
VALUES (%s, %s, %s)
""",
rows,
)
When the database and driver support multi-row inserts efficiently, a single statement can reduce round trips. Keep the values parameterized; constructing SQL by joining input strings turns a performance choice into a correctness and security defect. Also account for the database’s parameter and statement-size limits when the collection grows.
Choose an upsert only when the conflict behavior is part of the requirement. For PostgreSQL, ON CONFLICT states that behavior at the point where uniqueness is enforced:
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
ON CONFLICT (sku)
DO UPDATE SET quantity = EXCLUDED.quantity
""",
rows,
)
This replaces the stored quantity for each conflicting SKU. If the requirement is to add incoming stock instead, the update expression must say so. The distinction matters: a bulk operation can be fast and still silently encode the wrong business rule.
For very large PostgreSQL imports, consider COPY rather than generating a large insert statement. It is designed for moving many rows, but it has different constraints and does not provide ordinary per-row upsert behavior by itself. If imported rows need validation, deduplication, or conflict resolution, loading them into a staging table and merging from there may be the better operation.
When every target row receives the same change, keep the work in the database:
cursor.execute(
"""
UPDATE subscriptions
SET status = %s
WHERE expires_at < %s
""",
("expired", cutoff),
)
Sending one update per matching subscription wastes round trips and moves a set-based problem into application code. But when each row has a different value, avoid pretending that one uniform update will do. A CASE expression can handle a small, bounded set; for larger or more complex changes, a staging table joined to the target table is usually easier to inspect and maintain.
Deletes follow the same distinction. If all selected rows satisfy one condition, use one DELETE with that predicate. If the application already has a bounded list of identifiers, use a set-membership operation supported by the database or driver, and split the list when it exceeds parameter limits. The right choice depends on whether the database can express the selection itself or whether the application is supplying the exact rows to change.
Now loading...
Build operations that are clear and valid
A bulk statement should have one unambiguous meaning for every row. Define the required fields, their types, and what missing or NULL values mean before building the operation. A missing field must not accidentally become a request to overwrite a stored value with NULL, and a default must not be confused with an explicit value.
required = {"sku", "quantity"}
def validate_rows(rows):
validated = []
for row in rows:
if set(row) != required:
raise ValueError("Each row must contain exactly sku and quantity")
sku = row["sku"]
quantity = row["quantity"]
if not isinstance(sku, str) or not sku:
raise ValueError("sku must be a non-empty string")
if not isinstance(quantity, int) or quantity < 0:
raise ValueError("quantity must be a non-negative integer")
validated.append((sku, quantity))
return validated
Validate the collection as well as its individual rows. If the same key appears twice in one batch, decide which value wins or reject the batch. Leaving that behavior to statement order is fragile: databases may reject duplicate conflict keys within one upsert statement, while other execution strategies may apply the rows sequentially.
def reject_duplicate_skus(rows):
seen = set()
for sku, quantity in rows:
if sku in seen:
raise ValueError(f"Duplicate sku in batch: {sku}")
seen.add(sku)
Bind values as parameters, but do not expect parameters to stand in for table names, column names, or sort directions. Those are parts of the SQL structure, not values. If a query must choose among columns, map an application-level choice to a fixed, trusted SQL fragment rather than interpolating arbitrary input.
ORDER_COLUMNS = {
"sku": "sku",
"quantity": "quantity",
}
column = ORDER_COLUMNS.get(requested_order)
if column is None:
raise ValueError("Unsupported order column")
cursor.execute(
f"SELECT sku, quantity FROM inventory ORDER BY {column}"
)
Keep generated SQL aligned with the database constraints. An upsert conflict target must correspond to a unique constraint or index, and the update clause must preserve invariants that ordinary writes preserve. If a row is valid by itself but invalid in combination with another row, such as two entries claiming the same unique key, validate that relationship before execution or let a deliberate database constraint reject the batch. The operation is only clear when its behavior for every conflicting row is
Execute batches with deliberate error handling
A batch needs an explicit failure boundary. If its rows represent one business operation, execute them in one transaction and commit only when the entire operation succeeds. Otherwise, a failure on the last row can leave earlier rows committed while the caller believes the operation failed.
try:
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
""",
rows,
)
connection.commit()
except Exception:
connection.rollback()
raise
Do not catch an error merely to log it and continue. The caller needs to know that the write failed, and the transaction must be rolled back before the connection is reused. A broad exception handler is appropriate at this boundary only if it performs cleanup and re-raises; handle specific errors when the application can take a specific action.
Chunking controls memory use, statement size, and transaction duration, but it changes the failure contract. Committing after every chunk means earlier chunks remain stored if a later chunk fails. If all rows must succeed together, chunk the statements inside one transaction and commit once. If partial progress is acceptable, make that behavior explicit and return or persist enough information to resume safely.
def write_in_chunks(cursor, rows, chunk_size):
for start in range(0, len(rows), chunk_size):
chunk = rows[start:start + chunk_size]
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
""",
chunk,
)
try:
write_in_chunks(cursor, rows, chunk_size=500)
connection.commit()
except Exception:
connection.rollback()
raise
A malformed row is not a transient database failure. Do not retry it unchanged. If the contract requires all-or-nothing behavior, reject the whole batch and report the validation error. If valid rows may be accepted independently, isolate failures deliberately, for example with savepoints, and record which rows failed. Savepoint-based recovery adds database work and should not be used to conceal an unclear validation policy.
failed_rows = []
for index, row in enumerate(rows):
cursor.execute("SAVEPOINT row_write")
try:
cursor.execute(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
""",
row,
)
cursor.execute("RELEASE SAVEPOINT row_write")
except IntegrityError as error:
cursor.execute("ROLLBACK TO SAVEPOINT row_write")
cursor.execute("RELEASE SAVEPOINT row_write")
failed_rows.append((index, row, str(error)))
connection.commit()
Retry only errors that may succeed when the transaction is attempted again, such as a deadlock or a serialization failure. A uniqueness violation, invalid value, or missing required field will not be repaired by repeating the same statement. Retry the whole transaction when the database has aborted it; retrying only the last statement can reuse reads and decisions made in a transaction that’s no longer valid.
def is_retryable(error):
return getattr(error, "sqlstate", None) in {"40001", "40P01"}
def run_with_retry(operation, attempts=3):
for attempt in range(attempts):
try:
return operation()
except DatabaseError as error:
connection.rollback()
if not is_retryable(error) or attempt + 1 == attempts:
raise
Retries also require idempotent behavior. An insert that increments a balance, sends a notification, or creates an external side effect may apply that effect twice if the client loses the connection after the database commits but before the client receives confirmation. Use a stable request identifier with a uniqueness constraint, or otherwise make repeated execution produce the same stored result. Do not infer success or failure from a network exception alone; the commit may have completed even when its acknowledgment was lost.
Finally, treat affected-row counts as database- and driver-specific. Some drivers report an aggregate count for executemany, some report an unknown value, and upserts may count inserts and updates differently. When correctness depends on which rows changed, use a database operation that returns the affected identifiers or query the resulting state within the transaction, rather than assuming
Keep writes efficient without sacrificing correctness
that every driver reports the same meaning for a row count. When the write must prove which records changed, make that result part of the database operation. PostgreSQL can return identifiers from an update, avoiding a second query based on assumptions about driver bookkeeping:
cursor.execute(
"""
UPDATE inventory
SET quantity = incoming.quantity
FROM (VALUES %s) AS incoming(sku, quantity)
WHERE inventory.sku = incoming.sku
RETURNING inventory.sku
""",
values,
)
updated_skus = {row[0] for row in cursor.fetchall()}
The exact syntax for binding a set of values varies by driver. Use its supported bulk-binding mechanism; do not format a Python list into SQL text. If the database cannot return the needed information in the same statement, select the affected rows inside the transaction before committing.
Choose a chunk size that respects the database’s parameter limit and keeps each statement manageable. For a statement with three bound values per row and a limit of 60,000 parameters, the theoretical maximum is 20,000 rows; leave room for other parameters and practical statement-size limits. A smaller chunk can reduce memory pressure without changing the transaction boundary:
def chunks(items, size):
for start in range(0, len(items), size):
yield items[start:start + size]
try:
for batch in chunks(rows, 2_000):
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
ON CONFLICT (sku)
DO UPDATE SET quantity = EXCLUDED.quantity
""",
batch,
)
connection.commit()
except Exception:
connection.rollback()
raise
Chunking is a transport and resource-control choice, not permission to weaken the operation’s guarantees. Keep one transaction when the rows must become visible together. If the transaction is too large to hold locks or consume resources safely, define a resumable import protocol instead: give the job a stable identifier, record completed chunks, and make repeating a chunk produce the same database state.
Use constraints to protect invariants at the point of change. Application-side checks can improve error messages, but they cannot prevent two concurrent writers from both observing that a key is unused. A unique constraint can. Likewise, encode simple bounds and required values as database constraints so every write path, not only this batch function, obeys them.
CREATE TABLE inventory (
sku TEXT PRIMARY KEY,
quantity INTEGER NOT NULL CHECK (quantity >= 0),
version INTEGER NOT NULL DEFAULT 0
);
When an update depends on the value previously read by the application, include that value in the write predicate. This optimistic concurrency check prevents a batch from overwriting a change made after the read:
cursor.execute(
"""
UPDATE inventory
SET quantity = %s,
version = version + 1
WHERE sku = %s
AND version = %s
RETURNING sku
""",
(new_quantity, sku, expected_version),
)
if cursor.fetchone() is None:
raise ConcurrentUpdateError(sku)
For many rows, carry the expected version alongside each proposed value and perform the comparison in a set-based operation or through a staging table. Count or return the rows that matched; a missing match is a conflict to resolve, not a reason to silently retry with a newer version. If the rule is instead to add stock, express the addition atomically in SQL so concurrent writers do not overwrite one another’s increments:
cursor.execute(
"""
UPDATE inventory
SET quantity = quantity + %s
WHERE sku = %s
AND quantity + %s >= 0
RETURNING sku, quantity
""",
(delta, sku, delta),
)
Indexes also affect the cost of bulk writes. Every inserted or changed row may require updates to indexes, so add indexes for queries the system needs rather than indexing every column by habit. At the same time, do not remove a unique index that enforces correctness merely to accelerate an import. For a large load, measure the cost of the chosen indexes and constraints against the duration of the write and the guarantees the table must preserve
Test bulk behavior against real database outcomes
A mock can verify that a function called executemany. It cannot verify that the database enforced a unique constraint, that an upsert preserved the intended value, or that a failed transaction left no partial data behind. Exercise bulk writes against the database engine and driver used in production, with the relevant schema and constraints in place.
Give each test isolated data. A transaction this is rolled back after a test is useful for ordinary cases, but it cannot test behavior that must be observed across connections or after a commit. For those cases, use a disposable database or schema and remove it after the test.
def test_upsert_replaces_quantity(db):
with db.cursor() as cursor:
cursor.execute(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
""",
("A-17", 4),
)
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
ON CONFLICT (sku)
DO UPDATE SET quantity = EXCLUDED.quantity
""",
[("A-17", 9), ("B-42", 3)],
)
cursor.execute(
"""
SELECT sku, quantity
FROM inventory
WHERE sku IN (%s, %s)
ORDER BY sku
""",
("A-17", "B-42"),
)
result = cursor.fetchall()
assert result == [("A-17", 9), ("B-42", 3)]
Assert the stored state, not only that the call returned without raising. Include cases for a new key, a conflicting key, boundary values, and any distinction between omitted fields and explicit NULL values. If the operation promises to return changed identifiers, verify those identifiers against the rows actually changed.
Test the failure boundary with a constraint violation in the middle of a batch. If the contract is all-or-nothing, confirm that none of the batch’s writes remain visible after rollback:
def test_invalid_row_rolls_back_entire_batch(db):
rows = [
("A-17", 4),
("B-42", -1),
("C-09", 8),
]
with pytest.raises(DatabaseError):
with db.cursor() as cursor:
cursor.executemany(
"""
INSERT INTO inventory (sku, quantity)
VALUES (%s, %s)
""",
rows,
)
db.commit()
db.rollback()
with db.cursor() as cursor:
cursor.execute(
"""
SELECT sku
FROM inventory
WHERE sku IN (%s, %s, %s)
""",
("A-17", "B-42", "C-09"),
)
assert cursor.fetchall() == []
The precise exception class and transaction behavior depend on the driver. After an error, explicitly roll back before querying or reusing the connection; many databases mark the transaction as failed until that rollback occurs. Keep the test focused on the observable guarantee, while using the production driver to establish its actual exception behavior.
Do not use a single connection to test concurrent writes. A uniqueness race requires independent transactions, each with its own connection, so the database—not a shared cursor or application lock—decides which write succeeds. Coordinate the two writers with barriers or events, then assert the final row and the expected conflict outcome.
Also test the boundaries introduced by chunking. A batch that spans several chunks should produce the same final state as a batch that fits in one, and a failure in a later chunk should leave no earlier chunk committed when the contract is atomic. For resumable imports, run the same chunk twice and verify that the stable job or row identifier prevents duplicate effects.
Run these tests against the database version used in deployment. SQL syntax, conflict handling, parameter limits, isolation behavior, and driver row counts can differ between engines and versions; a test against an in-memory substitute can pass while production rejects the statement or reports different results.
Source: https://www.pythonfaq.net/how-to-run-bulk-operations-in-mongodb-via-pymongo-in-python/
