SQLAlchemy Session Leak — QueuePool Timeout in Production

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# The connection string tells SQLAlchemy which database to use.
# 'sqlite:///bookstore.db' creates a file called bookstore.db in the current directory.
# For PostgreSQL it would be: 'postgresql://user:password@localhost/dbname'
DATABASE_URL = "sqlite:///bookstore.db"

# Create the engine — this is the one-time setup that manages the connection pool.
# echo=True prints every SQL statement SQLAlchemy generates, great for debugging.
engine = create_engine(DATABASE_URL, echo=True)

# sessionmaker returns a class (a factory), not an instance.
# Every time you call SessionLocal(), you get a fresh, independent Session.
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# Base is the parent class all our ORM models will inherit from.
# It holds the metadata (table definitions) SQLAlchemy needs to create the schema.
Base = declarative_base()


# Helper function for safely opening and closing sessions.
# Use this as a context manager anywhere you need database access.
def get_db_session():
    session = SessionLocal()
    try:
        yield session          # Hand the session to the caller
        session.commit()       # Auto-commit if no exception was raised
    except Exception:
        session.rollback()     # Roll back ALL changes if anything went wrong
        raise
    finally:
        session.close()        # Always close the session to return connection to pool


print("Engine and session factory created successfully.")
print(f"Database URL: {DATABASE_URL}")

from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from datetime import datetime, timezone

# Import Base from our setup file
from database_setup import Base, engine


class Author(Base):
    """Represents an author in the bookstore database."""
    __tablename__ = "authors"   # Exact table name in the database

    # Integer primary key — SQLAlchemy auto-increments this for SQLite
    id = Column(Integer, primary_key=True, index=True)

    # nullable=False means this column is required — INSERT will fail without it
    name = Column(String(150), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    bio = Column(Text, nullable=True)   # Optional field

    # relationship() adds a Python-level attribute — NOT a database column.
    # 'books' will be a list of Book objects belonging to this author.
    # back_populates='author' means the Book model has a matching 'author' attribute.
    books = relationship("Book", back_populates="author", cascade="all, delete-orphan")

    def __repr__(self):
        # __repr__ makes debugging so much easier — you see useful info, not memory addresses
        return f"<Author(id={self.id}, name='{self.name}')>"


class Book(Base):
    """Represents a book in the bookstore database."""
    __tablename__ = "books"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(300), nullable=False)
    isbn = Column(String(13), unique=True, nullable=False)
    price = Column(Float, nullable=False)
    published_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    # ForeignKey creates the actual database constraint linking books.author_id to authors.id
    author_id = Column(Integer, ForeignKey("authors.id"), nullable=False)

    # This gives us book.author — a direct reference to the Author object
    author = relationship("Author", back_populates="books")

    def __repr__(self):
        return f"<Book(id={self.id}, title='{self.title}', price=${self.price})>"


# Create all tables defined above in the actual database file.
# If the tables already exist, this does nothing — it won't destroy existing data.
Base.metadata.create_all(bind=engine)
print("Tables created: 'authors' and 'books'")

from database_setup import SessionLocal
from models import Author, Book


def seed_bookstore_data():
    """Insert sample authors and books into the database."""
    with SessionLocal() as session:
        # --- INSERT: Create Author objects and add them to the session ---
        author_rowling = Author(
            name="J.K. Rowling",
            email="jk@rowling.com",
            bio="Author of the Harry Potter series."
        )
        author_martin = Author(
            name="George R.R. Martin",
            email="grrm@westeros.com",
            bio="Author of A Song of Ice and Fire."
        )

        # session.add_all() is more efficient than calling session.add() multiple times
        session.add_all([author_rowling, author_martin])

        # flush() writes changes to the DB within this transaction but doesn't commit yet.
        # We need this so author_rowling.id gets populated before we create the books.
        session.flush()

        print(f"Authors flushed — Rowling ID: {author_rowling.id}, Martin ID: {author_martin.id}")

        # --- INSERT: Create Book objects, linking them to authors via author_id ---
        books = [
            Book(title="Harry Potter and the Philosopher's Stone", isbn="9780747532699", price=12.99, author_id=author_rowling.id),
            Book(title="Harry Potter and the Chamber of Secrets", isbn="9780747538493", price=13.99, author_id=author_rowling.id),
            Book(title="A Game of Thrones", isbn="9780553103540", price=15.99, author_id=author_martin.id),
        ]
        session.add_all(books)
        session.commit()  # Now EVERYTHING above is written permanently to the database
        print("All data committed successfully.")


def query_bookstore_data():
    """Demonstrate various query and filter patterns."""
    with SessionLocal() as session:

        # --- QUERY ALL: Returns a list of all Author objects ---
        all_authors = session.query(Author).all()
        print(f"\nAll authors ({len(all_authors)} total):")
        for author in all_authors:
            print(f"  {author}")  # Uses our __repr__ method

        # --- FILTER BY: Simple equality filter, returns first match or None ---
        rowling = session.query(Author).filter_by(name="J.K. Rowling").first()
        print(f"\nFound author: {rowling}")

        # --- RELATIONSHIP ACCESS: Access books via the relationship attribute ---
        # SQLAlchemy issues a second query here automatically (lazy loading)
        print(f"\nRowling's books ({len(rowling.books)} total):")
        for book in rowling.books:
            print(f"  {book}")

        # --- FILTER with expression: Find all books priced above $13 ---
        # Book.price > 13.00 generates a SQL WHERE clause: WHERE books.price > 13.0
        expensive_books = session.query(Book).filter(Book.price > 13.00).all()
        print(f"\nBooks over $13.00:")
        for book in expensive_books:
            print(f"  {book.title} — ${book.price}")

        # --- UPDATE: Modify an attribute and commit ---
        cheap_book = session.query(Book).filter_by(isbn="9780747532699").one()
        cheap_book.price = 14.99  # SQLAlchemy detects this change automatically
        session.commit()
        print(f"\nPrice updated — {cheap_book.title} now costs ${cheap_book.price}")


if __name__ == "__main__":
    seed_bookstore_data()
    query_bookstore_data()

from sqlalchemy.orm import joinedload, subqueryload
from sqlalchemy import select
from database_setup import SessionLocal
from models import Author, Book


def demonstrate_n_plus_1_problem():
    """Shows what NOT to do — this fires one query per author."""
    print("=== N+1 Problem (BAD) ===")
    with SessionLocal() as session:
        # This fires ONE query: SELECT * FROM authors
        all_authors = session.query(Author).all()

        for author in all_authors:
            # EACH iteration fires ANOTHER query: SELECT * FROM books WHERE author_id = ?
            # With 1000 authors, that's 1001 total queries!
            book_count = len(author.books)
            print(f"  {author.name} has {book_count} book(s)")


def demonstrate_eager_loading():
    """Shows the right approach — loads everything in one query."""
    print("\n=== Eager Loading with joinedload (GOOD) ===")
    with SessionLocal() as session:
        # joinedload tells SQLAlchemy: 'when you fetch authors, also JOIN and fetch their books'
        # This produces ONE query with a JOIN instead of N+1 separate queries
        all_authors = (
            session.query(Author)
            .options(joinedload(Author.books))   # Pre-load the books relationship
            .all()
        )

        for author in all_authors:
            # author.books is already loaded — no extra database query fires here
            book_count = len(author.books)
            print(f"  {author.name} has {book_count} book(s)")
            for book in author.books:
                print(f"    -> {book.title} (${book.price})")


def find_authors_with_expensive_books():
    """Join across tables to filter parents based on child attributes."""
    print("\n=== Cross-Table Join Filter ===")
    with SessionLocal() as session:
        # .join(Book) tells SQLAlchemy to JOIN books on the foreign key relationship
        # .filter(Book.price > 15.00) then filters using the joined table's columns
        # .distinct() prevents the same author appearing multiple times if they have
        # multiple books matching the filter
        authors_with_pricey_books = (
            session.query(Author)
            .join(Book)                           # JOIN books ON authors.id = books.author_id
            .filter(Book.price > 15.00)           # WHERE books.price > 15.00
            .distinct()                           # Deduplicate if author has multiple matches
            .all()
        )

        print("Authors with at least one book over $15.00:")
        for author in authors_with_pricey_books:
            print(f"  {author.name}")


if __name__ == "__main__":
    demonstrate_n_plus_1_problem()
    demonstrate_eager_loading()
    find_authors_with_expensive_books()

from database_setup import SessionLocal
from models import Book


def safe_transfer_books(from_author_id, to_author_id, isbn_list):
    """
    Transfer ownership of multiple books from one author to another.
    If any step fails, all changes are rolled back.
    """
    with SessionLocal() as session:
        try:
            # Find source and target authors
            from_author = session.query(Author).get(from_author_id)
            to_author = session.query(Author).get(to_author_id)
            if not from_author or not to_author:
                raise ValueError("One or both authors not found")

            for isbn in isbn_list:
                book = session.query(Book).filter_by(isbn=isbn).first()
                if not book:
                    raise ValueError(f"Book with ISBN {isbn} not found")
                if book.author_id != from_author_id:
                    raise ValueError(f"Book {book.title} is not owned by source author")
                book.author_id = to_author_id  # This changes the author
            # If we reached here, success — commit
            session.commit()
            print(f"Transferred {len(isbn_list)} books successfully.")
        except Exception as e:
            session.rollback()
            print(f"Transaction rolled back due to error: {e}")
            raise  # Re-raise so caller knows it failed


# Example of how autoflush can cause surprise queries
# If autoflush=True (default), before session.query(Author).get(...)
# session flushes any pending changes, potentially writing partial data.
# Better to use autoflush=False for explicit control.

def demonstrate_autoflush_issue():
    with SessionLocal() as session:
        # This book object is 'pending' — not yet in database
        new_book = Book(title="Untitled", isbn="0000000000000", price=9.99, author_id=1)
        session.add(new_book)

        # Autoflush fires here! The 'Untitled' book gets inserted before the query
        existing_book = session.query(Book).filter_by(isbn="9780747532699").first()
        # If we then rollback, the 'Untitled' book is rolled back too
        session.rollback()

from sqlalchemy import select, func
from database_setup import SessionLocal
from models import Author, Book


def new_style_query_examples():
    with SessionLocal() as session:
        # Basic select: all authors
        stmt = select(Author).order_by(Author.name)
        result = session.execute(stmt)
        authors = result.scalars().all()
        print(f"Authors (2.0 style): {[a.name for a in authors]}")

        # Filter with WHERE
        stmt = select(Book).where(Book.price > 14.00).order_by(Book.price.desc())
        result = session.execute(stmt)
        books = result.scalars().all()
        print(f"Expensive books: {[b.title for b in books]}")

        # Aggregation: count books per author
        stmt = (
            select(Author.name, func.count(Book.id).label("book_count"))
            .join(Book, Author.id == Book.author_id)
            .group_by(Author.id)
            .order_by(Author.name)
        )
        result = session.execute(stmt)
        for row in result:
            print(f"{row.name}: {row.book_count} books")

        # Scalar for single value
        stmt = select(func.count(Book.id)).where(Book.price > 15.00)
        count = session.execute(stmt).scalar()
        print(f"Books over $15: {count}")

if __name__ == "__main__":
    new_style_query_examples()

// io.thecodeforge

from sqlalchemy import create_engine, text

# The engine manages a connection pool. Never create one per request.
engine = create_engine(
    "postgresql://app:secret@db:5432/production",
    pool_size=10,
    max_overflow=20,
    echo=True  # Set to False in prod. Use logging instead.
)

with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print(result.fetchone())

// io.thecodeforge

from sqlalchemy.orm import Session

# Correct: session as a context manager
with Session(engine) as session:
    session.begin()
    try:
        user = User(name="alice", email="alice@example.com")
        session.add(user)
        session.commit()
    except:
        session.rollback()
        raise
    # session closes automatically

# Wrong: manual open/close without guarantee
session = Session(engine)
try:
    user = session.get(User, 42)
    user.name = "bob"
    session.commit()
finally:
    session.close()  # at least this runs