SQLAlchemy upsert vs raw SQL for idempotent grant records

Problem statement #

You have a validated batch of grant records keyed on nih_project_id, and re-running the ingestion must update each existing row in place — never insert a duplicate and never lose the audit hash — but you have to choose how to write the upsert: SQLAlchemy’s dialect pg_insert(...).on_conflict_do_update(...), or a hand-written INSERT ... ON CONFLICT ... DO UPDATE. This guide compares the two on correctness, portability, audit-hash handling, performance, and safety, and gives a clear recommendation for when each wins.

It sits under Schema Validation Pipelines, part of Automated Ingestion & Data Sync Workflows. The validation gate has already partitioned the batch and handed you clean records — the sibling how-to Validating Grant Payloads with Pydantic v2 in Batch covers that step. This page is only about the write, and it inherits the idempotency contract set out in Grant Lifecycle Architecture Design: the same batch applied twice must leave the table byte-identical.

Prerequisites #

• Python 3.10+ with type hints and datetime.now(timezone.utc).
• Libraries: SQLAlchemy>=2.0 and psycopg[binary] for PostgreSQL. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • GRANT_DB_URL — the connection string, e.g. postgresql+psycopg://svc_grants:***@db.internal:5432/research.
• A unique constraint on nih_project_id. Both approaches depend on it; without the constraint there is no conflict target and no idempotency. Keep the DDL under version control with your University Policy Mapping Frameworks.

The two approaches side by side #

Both write the same row and compute the same audit_hash on the record’s canonical content, so a re-run reproduces the stored fingerprint exactly. The diagram frames the trade-off: the two paths converge on one table and one conflict target, differing only in who owns the ON CONFLICT clause — the ORM dialect layer, or you.

Figure: two write paths, one conflict target — the choice is who owns the ON CONFLICT clause and its parameter binding.

Approach A — SQLAlchemy dialect upsert #

The ORM path builds the statement from the dialect insert, then attaches on_conflict_do_update. The excluded pseudo-table exposes the values the conflicting insert would have written, so the update set references them symbolically rather than re-binding literals. Parameters are bound by the driver, and the conflict target is expressed as a column list, not raw text.

import hashlib
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import create_engine, Column, String, Numeric, DateTime
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy.dialects.postgresql import insert as pg_insert

Base = declarative_base()


class GrantRecord(Base):
    __tablename__ = "grant_records"
    nih_project_id = Column(String(50), primary_key=True)  # unique conflict target
    principal_investigator = Column(String(150), nullable=False)
    total_cost = Column(Numeric(14, 2), nullable=False)
    indirect_cost = Column(Numeric(14, 2), nullable=False)
    audit_hash = Column(String(64), nullable=False)
    updated_at = Column(DateTime, nullable=False)


def audit_hash(row: dict[str, Any]) -> str:
    """Deterministic SHA-256 over the content columns only (no timestamps)."""
    content = "|".join(
        f"{k}={row[k]}" for k in sorted(("nih_project_id", "principal_investigator",
                                         "total_cost", "indirect_cost"))
    )
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def upsert_sqlalchemy(engine, rows: list[dict[str, Any]]) -> int:
    now = datetime.now(timezone.utc)
    values = [
        {**r, "audit_hash": audit_hash(r), "updated_at": now}
        for r in rows
    ]
    stmt = pg_insert(GrantRecord).values(values)
    # excluded.* refers to the values the failed INSERT would have written.
    stmt = stmt.on_conflict_do_update(
        index_elements=["nih_project_id"],
        set_={
            "principal_investigator": stmt.excluded.principal_investigator,
            "total_cost": stmt.excluded.total_cost,
            "indirect_cost": stmt.excluded.indirect_cost,
            "audit_hash": stmt.excluded.audit_hash,
            "updated_at": stmt.excluded.updated_at,
        },
    )
    with Session(engine) as session:
        result = session.execute(stmt)
        session.commit()
        return result.rowcount

Approach B — hand-written raw SQL #

The raw path spells out the same statement as text and binds parameters with named placeholders. It uses PostgreSQL’s EXCLUDED keyword directly. Crucially, values are passed through executemany bind parameters — never string-formatted into the SQL — so the statement is immune to injection and the driver handles quoting and typing.

from sqlalchemy import text


UPSERT_SQL = text("""
    INSERT INTO grant_records
        (nih_project_id, principal_investigator, total_cost, indirect_cost,
         audit_hash, updated_at)
    VALUES
        (:nih_project_id, :principal_investigator, :total_cost, :indirect_cost,
         :audit_hash, :updated_at)
    ON CONFLICT (nih_project_id) DO UPDATE SET
        principal_investigator = EXCLUDED.principal_investigator,
        total_cost             = EXCLUDED.total_cost,
        indirect_cost          = EXCLUDED.indirect_cost,
        audit_hash             = EXCLUDED.audit_hash,
        updated_at             = EXCLUDED.updated_at
""")


def upsert_raw_sql(engine, rows: list[dict[str, Any]]) -> int:
    now = datetime.now(timezone.utc)
    params = [
        {**r, "audit_hash": audit_hash(r), "updated_at": now}
        for r in rows
    ]
    with engine.begin() as conn:
        # Named bind parameters — NEVER f-string the values into the SQL text.
        result = conn.execute(UPSERT_SQL, params)
        return result.rowcount

Both functions are idempotent: the second application of the same batch recomputes the same audit_hash, matches the existing row on nih_project_id, and overwrites it with identical content. No duplicate row appears, and the fingerprint column is stable across runs.

Decision matrix #

Recommendation #

Prefer the SQLAlchemy dialect upsert as the default for grant ingestion. Its symbolic excluded references keep the update set honest when the schema changes, its parameters are bound by construction so injection is not even reachable, and swapping dialects for a test SQLite database or a future migration is a one-line change rather than a query rewrite. For a batch ingestion pipeline that a mixed team of Python developers maintains, those properties matter more than a marginal per-statement speed difference.

Reach for raw SQL in three conditions: when a DBA owns the statement and needs it legible in plain SQL for review; when you need a Postgres feature the dialect layer does not cleanly expose (a partial-index conflict target with a complex WHERE, or DO UPDATE ... WHERE predicates); or when a profiler proves the statement-construction overhead is material on a very hot path. Even then, keep every value in a bound parameter — the moment a value is formatted into the SQL string, you have traded a tractable maintenance cost for an injection vulnerability.

For genuinely large loads, both approaches should be fed by the asynchronous commit path in Async Processing & Queue Management so a slow write never stalls validation.

Verification #

Confirm both paths are correct and interchangeable:

1. Identical row on re-run. Apply the same batch twice through either function. SELECT count(*) FROM grant_records WHERE nih_project_id = :pid must return 1, and the row must be byte-identical between runs except for updated_at.
2. Identical audit hash. Recompute audit_hash(row) from the persisted content columns and confirm it equals the stored audit_hash. Because the hash excludes timestamps, it must be stable across every re-run.
3. Cross-approach equivalence. Write a record with Approach A, then re-apply the identical record with Approach B. The row’s content columns and audit_hash must not change — proving the two paths are substitutable.
4. Injection probe. Pass a principal_investigator value of '); DROP TABLE grant_records; -- through the raw path. It must be stored verbatim as a string and the table must survive, confirming parameters are bound rather than interpolated.

Troubleshooting #

Three gotchas specific to choosing between these approaches:

• String-formatting parameters into raw SQL. The single most dangerous mistake. Any use of an f-string or %-formatting to place a value into the SQL text opens an injection hole and breaks typing for Decimal and datetime. Always use named bind parameters (:field) and pass values as a params dict or list; let the driver quote and type them.
• Dialect-specific ON CONFLICT syntax. ON CONFLICT is PostgreSQL and SQLite; MySQL uses ON DUPLICATE KEY UPDATE with a different VALUES()/EXCLUDED mechanism. Raw SQL hard-codes one dialect, so a database migration means rewriting the statement. The SQLAlchemy dialect insert isolates that difference behind a stable Python API.
• The excluded-column set drifting out of sync. When a new column is added, both approaches need it in the update set — but only the ORM path can be lint-checked against the model. In raw SQL a forgotten EXCLUDED.new_col line means the column is inserted but never updated on conflict, so re-runs silently keep the stale value. Review the update set against the schema whenever a column is added.

Frequently asked questions #

Related #

• Up to the parent topic: Schema Validation Pipelines
• Validating Grant Payloads with Pydantic v2 in Batch
• Grant Lifecycle Architecture Design
• Async Processing & Queue Management