SQLAlchemy Session Leak — QueuePool Timeout in Production

Every serious Python application eventually needs to store data. Whether you're building a REST API, a web app, or an internal tool, you'll hit a point where a dictionary just doesn't cut it and you need a real database. The instinct for many developers is to write raw SQL strings scattered across their codebase — and that works until it really doesn't. Maintenance becomes a nightmare, security holes (hello, SQL injection) creep in, and switching databases feels like a full rewrite.

SQLAlchemy solves this by giving you two powerful tools in one library. First, its Core layer lets you build and execute SQL expressions using Python objects — it's still SQL-flavored thinking, but type-safe and composable. Second, and more importantly for most projects, its ORM (Object-Relational Mapper) layer lets you define your database tables as Python classes and interact with rows as if they were plain Python objects. The database becomes an implementation detail, not the center of your universe.

By the end of this article, you'll know how to define database models as Python classes, create and manage a database session, insert and query records using both the ORM and filter expressions, and set up a one-to-many relationship between two tables. You'll also know the mistakes that trip up almost every developer the first time — and exactly how to avoid them.

Why Connection Pooling Is Not Optional

SQLAlchemy's QueuePool is a built-in connection pool that reuses database connections to avoid the overhead of establishing a new TCP connection for every request. By default, it maintains up to 5 connections in the pool, with a timeout of 30 seconds when all connections are checked out. When a request fails to acquire a connection within that timeout, it raises a TimeoutError — the classic 'QueuePool limit of size 5 overflow 10 reached' crash.

The pool works as a FIFO queue: connections are checked out from the pool, used, and returned. If a connection is never returned (a leak), the pool depletes. The overflow parameter allows up to 10 additional connections to be created temporarily, but once those are exhausted, new requests block until a connection is returned or the timeout expires. This is not a bug — it's a safety valve that prevents your database from being overwhelmed by runaway connections.

You must use connection pooling in any production web service. Without it, each request opens a new connection, consuming database resources and increasing latency by 10-50ms per request. QueuePool is the default for SQLAlchemy's create_engine() and is appropriate for most web applications with moderate concurrency. For high-throughput services, tune pool_size and max_overflow based on your database's max_connections and your traffic patterns.

Setting Up SQLAlchemy: The Engine and the Session Factory

Before you can talk to a database, SQLAlchemy needs two things: an Engine and a Session. Think of the Engine as the phone line — it knows the database's address and how to connect to it. The Session is a single phone call on that line — it's where your actual work happens, and it tracks every change you make until you decide to commit them.

The Engine is created once, at app startup, using a connection string. SQLAlchemy supports PostgreSQL, MySQL, SQLite, and more — you only change the connection string to switch. For development and learning, SQLite is perfect because it's a file-based database that requires zero setup.

The Session is created via a sessionmaker factory bound to your Engine. You never create Sessions manually in production — you use the factory. This separation matters: the Engine is a long-lived, shared, thread-safe object; Sessions are short-lived and should be opened and closed per request or per task.

The declarative_base() function creates a base class that all your ORM model classes inherit from. This base class is what registers your models with SQLAlchemy's metadata system, so it knows which Python class maps to which database table.

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}")

Defining ORM Models: Mapping Python Classes to Database Tables

An ORM model is a Python class that represents a database table. Each class attribute that uses SQLAlchemy's Column type maps to a column in that table. When you define a model, you're doing two things at once: describing the database schema AND defining the Python object you'll work with in your code.

The __tablename__ attribute is mandatory — it tells SQLAlchemy the exact name of the table in the database. Column types like Integer, String, and DateTime are SQLAlchemy's type system, which maps to the appropriate native type for whatever database you're using.

Relationships between tables are defined using relationship(), which tells SQLAlchemy how two models connect logically. The ForeignKey defines the database-level link, while relationship() adds the Python-level convenience — letting you access related objects directly as attributes instead of writing join queries manually.

Once your models are defined, calling Base.metadata.create_all(engine) inspects all classes that inherit from Base and creates the corresponding tables in the database if they don't already exist. It's safe to call repeatedly — it won't overwrite existing tables.

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'")

Sessions in Action: Inserting, Querying, and Filtering Records

The Session is where all the action happens. Think of it as a scratchpad — you add objects to it, modify them, and delete them, and SQLAlchemy tracks every change. Nothing actually touches the database until you call session.commit(). This is called the Unit of Work pattern, and it's one of SQLAlchemy's most powerful ideas.

When you add an object with session.add(), it moves into a 'pending' state — tracked by the session but not yet written. After commit(), it moves to 'persistent' — it exists in the database AND in the session's identity map. If you query for it again, SQLAlchemy returns the same Python object from memory, not a fresh copy from the database.

Querying uses the session.query() method (classic ORM style) or the newer select() construct. Both work, but the newer select() style is the direction SQLAlchemy 2.0 is heading. Filters work like WHERE clauses — filter_by() accepts keyword arguments for simple equality checks, while filter() accepts more expressive comparison expressions for anything complex like greater-than, LIKE, or IN.

Understanding the difference between .all(), .first(), and .one() matters: .all() returns a list (empty if no results), .first() returns the first result or None, and .one() raises an exception if the result count isn't exactly one — useful when you absolutely expect a unique record.

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

Joins and Eager Loading: Avoiding the N+1 Query Problem

The N+1 query problem is the most common performance mistake developers make with ORMs, and SQLAlchemy's lazy loading makes it easy to fall into. Here's how it happens: you fetch 100 authors with one query, then loop over them accessing author.books — and SQLAlchemy fires 100 separate queries to load each author's books. One query becomes 101. That's N+1.

The fix is eager loading — telling SQLAlchemy to fetch the related data upfront in the same query. SQLAlchemy offers two main strategies: joinedload() uses a SQL JOIN to get everything in one query, and subqueryload() uses a second optimized query to load all related records at once. Both avoid N+1; which one to use depends on your data shape.

joinedload() is ideal when each parent has a small number of related children — the join stays manageable. subqueryload() is better when you have many parents each with many children, because joining would create a large Cartesian product that duplicates the parent rows. For most one-to-many relationships with moderate data, joinedload() is your default.

You can also write explicit joins using .join() and filter across tables — essential when you need to filter the parent based on a child's column, like finding all authors who have at least one book priced over $15.

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

Transactions, Rollbacks, and Session Boundaries

The session's transaction is what keeps your data consistent. When you call session.commit(), all pending changes are written atomically. If any part of the operation fails, you call session.rollback() to undo everything since the last commit — no partial writes, no data corruption.

But here's the thing: if an exception occurs inside a session and you don't explicitly rollback, the session stays in a 'defunct' state. Any further operation on that session raises an error. That's why the generator pattern in the setup section catches all exceptions, calls rollback, then re-raises — it prevents the session from being left in a broken state.

Session boundaries are critical when integrating with web frameworks. Open the session at the start of a request, commit at the end if successful, rollback on error, and always close. Most frameworks (Flask, FastAPI, Django REST) have middleware or dependency injection to manage this automatically — use them.

Be aware that autoflush=True (the default) automatically flushes pending changes before any query. This can cause surprising commits midway through a transaction. Disable autoflush (autoflush=False) when you need explicit control over when data hits the database.

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

Querying with the SQLAlchemy 2.0 Style: select() and Executable

SQLAlchemy 2.0 introduced a cleaner, more consistent way to build queries using the select() function. Instead of session.query(Author).filter(...), you write select(Author).where(...). The new style feels more like SQL but stays Pythonic, and it unifies the Core and ORM interfaces.

With the 2.0 style, you pass the result to session.execute() and extract scalars with .scalars().all() or .scalar(). This might feel like extra verbosity at first, but it pays off when you mix ORM objects and Core constructs in the same query. The new style also enforces explicit execution, making lazy query evaluation less surprising.

You can chain .where(), .order_by(), .limit(), and .offset() just like the old style. Aggregations use func.count(), func.sum(), and you group with .group_by(). The output is a Result object that you iterate over or convert to a list.

The 2.0 style is the future. SQLAlchemy 2.0 still supports the old style for backward compatibility, but new projects should adopt select() from day one.

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

What SQLAlchemy Actually Is (And Why You Should Care)

SQLAlchemy is not magic. It's a SQL toolkit and Object-Relational Mapper (ORM) that translates Python objects into database rows and back. The core value? You write Python, not raw SQL strings. When you push to production and the DBA swaps MySQL for PostgreSQL, your queries still work. The ORM handles the dialect translation. But here's the trap: many devs treat it as a black box. They never look at the generated SQL. That's how you get queries that pull 10,000 rows when you only need five. SQLAlchemy gives you two layers: Core (raw SQL expressions, no ORM magic) and ORM (full object mapping). Start with ORM, but learn Core when the ORM fights you. Use it because it enforces parameterized queries by default—bye-bye SQL injection. Use it because connection pooling is built in. Use it because your junior can read a Python class and understand the schema without digging through migration files. Do not use it if you need raw performance on simple CRUD—raw DBAPI is faster. But for 95% of applications, SQLAlchemy wins.

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

Why You Must Understand the Session Lifecycle

The session is the star of SQLAlchemy's ORM. It tracks changes to your objects and flushes them to the database. But here's what burns juniors: sessions are not thread-safe. Create one per request or per logical unit of work. Never share a session across threads. The session uses a pattern called 'identity map'. It keeps a cache of objects by primary key. Read the same row twice? You get the same Python object. That means changes in one part of your code are visible in another without a database round-trip. Sounds great until you forget you mutated an object and commit unintended changes. Rule: treat each session as a transaction boundary. Open it, do work, commit or rollback, close it. The most common production incident I've seen? Leaked sessions. A session opened, never closed, pool exhausted, app dead. Use a context manager (session.begin()) or try/finally. Or let FastAPI/Flask inject session scopes. Do not reinvent that wheel. Session lifecycle mismanagement causes the N+1 problem, stale reads, and deadlocks. Get it right before writing a single query.

// 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

SQLAlchemy 2.x: New Core and ORM Patterns

SQLAlchemy 2.0 introduced a unified API for both Core and ORM, emphasizing the select() construct and native support for asynchronous drivers. The ORM now uses a single Session.execute() method for all queries, returning Result objects. Key changes include: removal of Session.query(), mandatory use of select() with session.execute(), and improved type hints. For example, instead of session.query(User).filter(User.name == 'Alice').all(), you write:

``python from sqlalchemy import select stmt = select(User).where(User.name == 'Alice') result = session.execute(stmt) users = result.scalars().all() ``

This pattern works identically for Core and ORM, reducing cognitive overhead. Additionally, 2.0 introduces mapped_column() for declarative models, replacing the legacy Column() with better typing. The Session now supports session.get() for primary key lookups, and session.refresh() for re-fetching data. For bulk operations, use session.execute() with insert() or update() constructs. The new style also enforces explicit commit() or rollback() via context managers, preventing session leaks. Migrating to 2.0 is straightforward: replace query() calls with select(), use scalars() for ORM objects, and adopt mapped_column() in models. This section demonstrates the modern patterns that align with the article's focus on production reliability.

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column

engine = create_engine("sqlite:///example.db")

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str]

Base.metadata.create_all(engine)

with Session(engine) as session:
    # Insert using 2.0 style
    session.add(User(name="Alice", email="alice@example.com"))
    session.commit()

    # Query using select()
    stmt = select(User).where(User.name == "Alice")
    result = session.execute(stmt)
    user = result.scalars().one()
    print(user.name)

Async SQLAlchemy with asyncio and asyncpg

For high-concurrency applications, async SQLAlchemy with asyncio and asyncpg (PostgreSQL) provides non-blocking database access. SQLAlchemy 1.4+ supports async via AsyncEngine, AsyncSession, and async drivers. The async ORM mirrors the sync API but uses await for all I/O operations. Example setup:

```python from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") AsyncSessionLocal = sessionmaker(async_engine, class_=AsyncSession) ```

Queries use await session.execute(select(...)) and await session.commit(). The async session must be used within an async context manager or with async with. For example:

``python async def get_user(user_id: int): async with AsyncSessionLocal() as session: result = await session.execute(select(User).where(User.id == user_id)) return result.scalar_one_or_none() ``

Async SQLAlchemy works with asyncio.gather() for concurrent queries, but beware of connection pool limits. Use asyncpg for PostgreSQL; for other databases, use aiosqlite (SQLite) or aiomysql (MySQL). The async engine uses a separate pool that must be configured with pool_size and max_overflow to avoid timeouts. This section integrates with the article's theme of session leaks: async sessions are still bound to transactions and must be closed properly. Always use async with or await session.close() to release connections back to the pool.

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/testdb", echo=True)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def main():
    async with AsyncSessionLocal() as session:
        # Insert
        session.add(User(name="Bob", email="bob@example.com"))
        await session.commit()

        # Query
        stmt = select(User).where(User.name == "Bob")
        result = await session.execute(stmt)
        user = result.scalar_one()
        print(user.name)

asyncio.run(main())

Alembic: Database Migration Management

Alembic is the recommended migration tool for SQLAlchemy, enabling version-controlled schema changes. It generates migration scripts automatically by comparing your ORM models to the current database state. Setup involves initializing Alembic (alembic init alembic), configuring alembic.ini with the database URL, and editing env.py to import your Base metadata. Example env.py:

``python from myapp.models import Base target_metadata = Base.metadata ``

Then create a migration with alembic revision --autogenerate -m "add users table". This produces a script with upgrade() and downgrade() functions. For example:

```python def upgrade(): op.create_table('users', sa.Column('id', sa.Integer(), nullable=False), sa.Column('name', sa.String(), nullable=False), sa.PrimaryKeyConstraint('id') )

def downgrade(): op.drop_table('users') ```

Apply migrations with alembic upgrade head. Alembic tracks applied migrations in a table (alembic_version). Best practices: always review autogenerated scripts, test downgrades, and use --sql for offline migrations. In production, run migrations as part of deployment (e.g., in a Kubernetes init container). Alembic integrates with SQLAlchemy's session lifecycle: migrations run outside of application sessions, so they don't interfere with connection pooling. This section complements the article by showing how to manage schema changes without manual SQL, reducing the risk of session leaks from mismatched schemas.

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from myapp.models import Base  # Import your models' Base

config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

run_migrations_online()