SQLAlchemy Basics: Models, Queries, and Sessions Explained

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()