Getting IDs Between Queries in a Single SQLAlchemy Transaction

Case Study: Double Entry Accounting System
Why Are Database Transactions Important?
Challenge: Getting IDs Between Queries
Implementation with SQLAlchemy
Best Practices
Conclusion

Have you ever encountered a situation where you need to update more than one table, but all operations must be treated as a single unit? The process must either succeed completely or fail completely (rollback). There should be no condition where one table is successfully updated while another fails.

Case Study: Double Entry Accounting System

Let’s take a real example from an accounting system using double entry. Imagine you’re building a financial application where every transaction must be recorded on two sides: debit and credit. The basic accounting rule is: total debit must equal total credit.

For example, when a company purchases equipment worth $10,000:

Debit: Equipment Account (+$10,000)
Credit: Cash Account (-$10,000)

Imagine if the first insert (debit) succeeds, but the second insert (credit) fails due to a database error or network issue. The financial report will become unbalanced and the data will be inconsistent!

Why Are Database Transactions Important?

Database transactions provide ACID guarantees:

Atomicity: All operations succeed or all fail
Consistency: Data remains consistent according to business rules
Isolation: Transactions don’t interfere with each other
Durability: Committed changes are permanent

Without transactions, your application is vulnerable to data corruption and inconsistency.

Challenge: Getting IDs Between Queries

Now, here’s an interesting challenge: what if we need to get the ID from the first insert to use in the next insert, all within a single transaction?

Example scenario:

Insert transaction_header data (get transaction ID)
Insert transaction_details data using ID from step 1
Insert journal_entries data using the same ID
If any step fails, everything must be rolled back

Implementation with SQLAlchemy

Let’s see how to solve this problem with SQLAlchemy!

Model Setup

First, let’s define our database models:

from sqlalchemy import Column, Integer, String, Numeric, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime

Base = declarative_base()

class TransactionHeader(Base):
    __tablename__ = 'transaction_headers'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    transaction_date = Column(DateTime, default=datetime.utcnow)
    description = Column(String(200))
    total_amount = Column(Numeric(15, 2))
    status = Column(String(20), default='pending')
    
    details = relationship("TransactionDetail", back_populates="header")
    journal_entries = relationship("JournalEntry", back_populates="transaction")

class TransactionDetail(Base):
    __tablename__ = 'transaction_details'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    transaction_id = Column(Integer, ForeignKey('transaction_headers.id'))
    item_name = Column(String(100))
    quantity = Column(Integer)
    unit_price = Column(Numeric(15, 2))
    subtotal = Column(Numeric(15, 2))
    
    header = relationship("TransactionHeader", back_populates="details")

class JournalEntry(Base):
    __tablename__ = 'journal_entries'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    transaction_id = Column(Integer, ForeignKey('transaction_headers.id'))
    account_code = Column(String(20))
    account_name = Column(String(100))
    debit = Column(Numeric(15, 2), default=0)
    credit = Column(Numeric(15, 2), default=0)
    
    transaction = relationship("TransactionHeader", back_populates="journal_entries")

Using Session Transaction

This is the most common and recommended approach:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://user:password@localhost/mydb')
Session = sessionmaker(bind=engine)

def create_complete_transaction(data):
    session = Session()
    
    try:
        # 1. Insert header and get its ID
        header = TransactionHeader(
            description=data['description'],
            total_amount=data['total_amount']
        )
        session.add(header)
        session.flush()  # This is important! Flush generates ID without commit
        
        # Now header.id is available!
        transaction_id = header.id
        print(f"Transaction ID: {transaction_id}")
        
        # 2. Insert detail items using transaction_id
        for item in data['items']:
            detail = TransactionDetail(
                transaction_id=transaction_id,
                item_name=item['name'],
                quantity=item['quantity'],
                unit_price=item['price'],
                subtotal=item['quantity'] * item['price']
            )
            session.add(detail)
        
        # 3. Insert journal entries (double entry)
        # Debit: Inventory/Asset
        journal_debit = JournalEntry(
            transaction_id=transaction_id,
            account_code='1100',
            account_name='Inventory',
            debit=data['total_amount'],
            credit=0
        )
        session.add(journal_debit)
        
        # Credit: Cash
        journal_credit = JournalEntry(
            transaction_id=transaction_id,
            account_code='1000',
            account_name='Cash',
            debit=0,
            credit=data['total_amount']
        )
        session.add(journal_credit)
        
        # Validate double entry
        total_debit = data['total_amount']
        total_credit = data['total_amount']
        
        if total_debit != total_credit:
            raise ValueError("Debit and Credit must be equal!")
        
        # Commit all changes
        session.commit()
        print("Transaction completed successfully!")
        return transaction_id
        
    except Exception as e:
        # If there's an error, rollback all changes
        session.rollback()
        print(f"Transaction failed: {str(e)}")
        raise
    finally:
        session.close()

# Usage example
transaction_data = {
    'description': 'Purchase Office Equipment',
    'total_amount': 10000000,
    'items': [
        {'name': 'Laptop', 'quantity': 2, 'price': 4000000},
        {'name': 'Monitor', 'quantity': 4, 'price': 500000}
    ]
}

try:
    txn_id = create_complete_transaction(transaction_data)
    print(f"Created transaction with ID: {txn_id}")
except Exception as e:
    print(f"Failed to create transaction: {e}")

Key Difference: flush() vs commit()

The difference between flush() and commit() is crucial:

flush(): Sends SQL to the database and generates ID, but hasn’t committed yet. Still within the same transaction and can be rolled back.
commit(): Permanently saves all changes to the database.

With flush(), you can get the header.id that’s already generated by the database, but the transaction is still open. If there’s an error in the next steps, everything can be rolled back.

Using Context Manager

A cleaner approach with context manager:

from contextlib import contextmanager

@contextmanager
def transaction_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
def create_transaction_v2(data):
    with transaction_scope() as session:
        header = TransactionHeader(
            description=data['description'],
            total_amount=data['total_amount']
        )
        session.add(header)
        session.flush()
        
        # Use header.id for subsequent operations
        for item in data['items']:
            detail = TransactionDetail(
                transaction_id=header.id,
                item_name=item['name'],
                quantity=item['quantity'],
                unit_price=item['price'],
                subtotal=item['quantity'] * item['price']
            )
            session.add(detail)
        
        # Auto commit if no error
        # Auto rollback if exception occurs
        return header.id

Best Practices

Always use transactions for multi-table operations: Never insert into multiple tables without a transaction.
Use flush() to get IDs: Don’t commit in the middle of the process if there are still other operations.
Validate before commit: Ensure all business rules are met before committing.
Handle exceptions properly: Always catch exceptions and perform rollback.
Use context managers: Cleaner and automatically handles commit/rollback.
Test rollback scenarios: Ensure rollback works properly when errors occur.

Conclusion

Properly managing database transactions is a fundamental skill that every developer must master. With SQLAlchemy, you can easily:

Manage multi-table transactions
Get IDs from the first insert to use in subsequent queries
Ensure data consistency with ACID properties

Remember: use flush() to get IDs within the same transaction, and make sure all operations succeed before performing commit().

Happy coding! 🚀

Related Articles:

Table of Contents