Complete Guide to SQLAlchemy Configuration for Python Web Applications
Typical Problem with Sessions
FastAPI developers often encounter a situation: the app works locally, but in production after 10 minutes — an SSL SYSCALL error or BrokenPipeError. The cause is that the connection pool contains dead sockets. SQLAlchemy 2.0 with the pool_pre_ping option solves this, but proper configuration is only part of the path. Without correct setup of async sessions and migrations, you risk N+1 queries and MissingGreenlet errors under load.
We have been configuring SQLAlchemy for Python web applications on FastAPI and Flask for over five years, completing over 50 projects. During this time we've collected a set of best practices that guarantee stability even at 1500+ requests per second. In this article we'll break down key components: from async session to automatic Alembic migrations. The SQLAlchemy 2.0 Documentation recommends exactly this approach.
For example, in one project with a peak load of 2000 RPS we encountered TimeoutError due to the lack of pool_pre_ping. After implementing this option and increasing the pool to 30 connections, response time dropped by 40%. Such results are only possible with correct configuration of the entire chain.
How to Configure an Async Session for FastAPI (Step-by-Step)
- Install asyncpg and SQLAlchemy:
pip install asyncpg sqlalchemy[asyncio]. - Create engine with
create_async_engine:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/mydb" engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20, pool_pre_ping=True, echo=False) AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) class Base(DeclarativeBase): pass pool_pre_ping=True checks the connection before using it — mandatory for production. Without it, dead connections cause 500 errors, especially in cloud environments with long timeouts. Additionally, set pool_recycle to 3600 seconds for automatic replacement of old connections.
- Inject the session via dependency injection: create a
get_dbdependency that opens a session, performs commit or rollback.
Why expire_on_commit=False is Critical for Async
By default after commit() SQLAlchemy expires all objects. Accessing attributes in async mode causes MissingGreenlet. Disabling it keeps objects available without extra queries. This boosts performance by about 20% and eliminates many debug sessions.
Models and Queries in 2.0 Style
The new typed API: Mapped + mapped_column instead of the old Column. Example of a user model with a relationship:
from datetime import datetime from typing import Optional from sqlalchemy import String, Enum, func from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base import enum class UserRole(enum.Enum): admin = "admin" editor = "editor" viewer = "viewer" class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False) password_hash: Mapped[str] = mapped_column(String(255), nullable=False) role: Mapped[UserRole] = mapped_column(Enum(UserRole), default=UserRole.viewer, nullable=False) created_at: Mapped[datetime] = mapped_column(server_default=func.now(), nullable=False) updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now(), nullable=False) posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin") lazy="selectin" — a safe strategy for async: executes a separate SELECT ... WHERE id IN (...), no MissingGreenlet. Compared to joinedload, it avoids giant JOINs, giving a performance boost up to 30% on queries with many relationships.
Queries:
from sqlalchemy import select from app.models.user import User from app.models.post import Post async def get_published_posts_with_authors(db: AsyncSession, limit: int = 20, offset: int = 0) -> list[Post]: stmt = select(Post).join(Post.author).where(Post.status == "published").order_by(Post.created_at.desc()).limit(limit).offset(offset) result = await db.execute(stmt) return list(result.scalars().all()) Transactions and Migrations
For isolation of operations use nested transactions: async with db.begin_nested():. This is convenient for rolling back individual operations without reverting the entire transaction.
Configuring Alembic for async: Initialization:
alembic init -t async alembic Edit alembic/env.py:
from logging.config import fileConfig from sqlalchemy.ext.asyncio import async_engine_from_config from alembic import context from app.database import Base import app.models # noqa: F401 config = context.config fileConfig(config.config_file_name) target_metadata = Base.metadata def run_migrations_online(): connectable = async_engine_from_config(config.get_section(config.config_ini_section), prefix="sqlalchemy.") async def do_run(): async with connectable.connect() as connection: await connection.run_sync(context.configure, connection=connection, target_metadata=target_metadata, compare_type=True) async with context.begin_transaction(): await connection.run_sync(context.run_migrations) import asyncio asyncio.run(do_run()) run_migrations_online() compare_type=True — Alembic will track type changes. This saves time during refactoring.
Common Errors from Incorrect Configuration
Click to expand common errors
- MissingGreenlet — from lazy loading in async. Solution: use
lazy='selectin'orawait db.refresh(). - N+1 queries — especially dangerous in async. Use
selectinloadorjoinedload. - Connection timeouts — solved via
pool_pre_pingandpool_recycle. - Data race — transactions must be idempotent. Our engineers check this at code review stage.
Comparison of Sync and Async Approaches
| Criterion | Synchronous | Asynchronous |
|---|---|---|
| Driver | psycopg2 | asyncpg |
| Engine | create_engine | create_async_engine |
| Session | sessionmaker | async_sessionmaker |
| Queries | session.execute | await db.execute |
| Throughput | ~500 req/s | ~1500 req/s |
Async approach gives a 3x increase in requests per second, which is critical for high-load projects. This makes async 3 times better than sync for high-traffic applications.
What's Included in the Work
- Audit of current SQLAlchemy configuration and identification of bottlenecks.
- Setup of async session with
pool_pre_ping, optimization of connection pool. - Design of models with correct lazy strategies and typing.
- Implementation of Alembic migrations with autogeneration and type control.
- Integration of session into FastAPI/Flask via dependency injection.
- Operations documentation and deployment instructions.
- Post-deployment support: 2 weeks of consultations.
Timelines and Cost
We provide a turnkey solution: setting up SQLAlchemy from scratch for a new project takes from 1 business day, costing from $500. Migrating an existing application from 1.4 to 2.0 — from 2 days, priced from $1,200. Cost is calculated individually after assessing the volume of models and queries.
Get a consultation for your project — our specialists will help configure SQLAlchemy to avoid problems under load. Order an audit of your current configuration and receive specific recommendations for performance improvement. We guarantee production reliability with 5+ years of experience and over 50 successful projects.







