Savepoints

Savepoints allow partial rollbacks within a transaction.

Basic Savepoint Usage

conn = psycopg2.connect(...)
cur = conn.cursor()

cur.execute("INSERT INTO users (name) VALUES ('Alice')")

# Create savepoint
cur.execute("SAVEPOINT sp1")

cur.execute("INSERT INTO users (name) VALUES ('Bob')")

# Something went wrong with Bob, rollback to savepoint
cur.execute("ROLLBACK TO SAVEPOINT sp1")

# Alice insert is preserved, Bob is rolled back
cur.execute("INSERT INTO users (name) VALUES ('Charlie')")

conn.commit()  # Commits Alice and Charlie

Savepoints in Error Handling

def insert_with_retry(conn, data_list):
    """Insert multiple items, continue on individual failures"""
    cur = conn.cursor()
    successful = []
    failed = []

    for item in data_list:
        try:
            cur.execute("SAVEPOINT item_insert")
            cur.execute(
                "INSERT INTO items (name, value) VALUES (%s, %s)",
                (item['name'], item['value'])
            )
            cur.execute("RELEASE SAVEPOINT item_insert")
            successful.append(item)
        except psycopg2.Error as e:
            cur.execute("ROLLBACK TO SAVEPOINT item_insert")
            failed.append({'item': item, 'error': str(e)})

    conn.commit()
    return successful, failed

Using with Statement for Savepoints

from contextlib import contextmanager

@contextmanager
def savepoint(conn, name):
    """Context manager for savepoints"""
    cur = conn.cursor()
    cur.execute(f"SAVEPOINT {name}")
    try:
        yield
        cur.execute(f"RELEASE SAVEPOINT {name}")
    except Exception:
        cur.execute(f"ROLLBACK TO SAVEPOINT {name}")
        raise

# Usage
with savepoint(conn, "my_savepoint"):
    cur.execute("INSERT INTO users (name) VALUES ('Test')")
    # If exception occurs, rolls back to savepoint