Async SQLAlchemy: asyncpg, Alembic, and Production Patterns

In modern web development, asynchronous programming is essential for building high-performance applications that can handle thousands of concurrent connections. Python's asyncio ecosystem, combined with async database drivers like asyncpg, enables non-blocking database operations. SQLAlchemy, the leading ORM for Python, now offers full async support, allowing you to use the same powerful ORM patterns in async contexts.

This tutorial dives deep into async SQLAlchemy with asyncpg, covering setup, query patterns, Alembic migrations, and production-ready practices. You'll learn how to configure an async engine, manage sessions, handle transactions, and avoid common pitfalls. We'll also explore real-world scenarios like connection pooling, retry logic, and debugging async database issues.

By the end, you'll be equipped to build scalable async applications that interact with PostgreSQL efficiently, using SQLAlchemy's familiar API in a non-blocking way. Whether you're building a FastAPI service, an aiohttp app, or any asyncio-based system, these patterns will help you write robust, maintainable database code.

Setting Up Async SQLAlchemy with asyncpg

To get started, install the required packages: pip install sqlalchemy[asyncio] asyncpg. The async engine is created using create_async_engine instead of the sync create_engine. This engine uses asyncpg under the hood to communicate with PostgreSQL asynchronously.

Here's a basic setup:

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

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=5, max_overflow=10)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) ```

Always use async with for sessions to ensure proper cleanup:

``python async with async_session() as session: result = await session.execute(select(User).where(User.id == 1)) user = result.scalar_one() ``

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/testdb"

engine = create_async_engine(DATABASE_URL, echo=True, pool_size=5, max_overflow=10)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

async def get_user(user_id: int):
    async with async_session() as session:
        result = await session.execute(select(User).where(User.id == user_id))
        return result.scalar_one()

Defining Models and Running Queries

Models are defined similarly to sync SQLAlchemy, using declarative_base. However, all database operations must be awaited. Use select() from SQLAlchemy's core to build queries.

Example of a simple query:

```python from sqlalchemy import select

async def get_users(): async with async_session() as session: stmt = select(User).where(User.name.like('A%')) result = await session.execute(stmt) users = result.scalars().all() return users ```

For relationships, use selectinload to eagerly load related objects and avoid N+1 queries:

```python from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.posts)) result = await session.execute(stmt) ```

Async SQLAlchemy also supports raw SQL via text():

```python from sqlalchemy import text

result = await session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1}) ```

Remember that all operations that touch the database must be awaited, including commit(), rollback(), flush(), and refresh().

from sqlalchemy import select, text
from sqlalchemy.orm import selectinload

async def get_user_with_posts(user_id: int):
    async with async_session() as session:
        stmt = select(User).options(selectinload(User.posts)).where(User.id == user_id)
        result = await session.execute(stmt)
        return result.scalar_one()

async def create_user(name: str):
    async with async_session() as session:
        user = User(name=name)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

async def raw_query():
    async with async_session() as session:
        result = await session.execute(text("SELECT count(*) FROM users"))
        count = result.scalar()
        return count

Transactions and Session Management

Async SQLAlchemy provides two ways to manage transactions: automatic commit on context exit and explicit transaction blocks.

Automatic commit: When using async with async_session() as session:, the session will automatically commit if no exception occurs, otherwise it rolls back. This is the recommended pattern for simple operations.

Explicit transaction: Use async with session.begin(): to control the transaction boundaries manually. This is useful when you need to perform multiple operations in a single transaction:

``python async with async_session() as session: async with session.begin(): user = User(name='Alice') session.add(user) # other operations # transaction is committed here ``

You can also use await session.commit() and await session.rollback() explicitly, but be careful to always close the session.

Nested transactions are supported via savepoints:

``python async with session.begin_nested(): # this is a savepoint session.add(some_object) ``

In production, always handle exceptions and ensure sessions are closed. Use try/finally or context managers.

from sqlalchemy.exc import SQLAlchemyError

async def transfer_funds(sender_id: int, receiver_id: int, amount: float):
    async with async_session() as session:
        try:
            async with session.begin():
                sender = await session.get(User, sender_id)
                receiver = await session.get(User, receiver_id)
                sender.balance -= amount
                receiver.balance += amount
        except SQLAlchemyError as e:
            print(f"Transaction failed: {e}")
            raise
        # session is automatically closed

async def explicit_commit():
    session = async_session()
    try:
        user = User(name='Bob')
        session.add(user)
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

Alembic Migrations for Async Databases

Alembic supports async migrations with a few configuration changes. First, install alembic: pip install alembic. Then initialize: alembic init alembic.

Modify alembic/env.py to use async engine:

```python from sqlalchemy.ext.asyncio import create_async_engine from alembic import context

async def run_migrations_online(): connectable = create_async_engine(DATABASE_URL) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) await connectable.dispose()

def do_run_migrations(connection): context.configure(connection=connection, target_metadata=Base.metadata) with context.begin_transaction(): context.run_migrations()

def run_migrations_online(): asyncio.run(run_migrations_online()) ```

Note: The run_async() method is available in newer Alembic versions. Alternatively, use asyncio.run() as shown.

For offline mode, you can generate SQL scripts synchronously.

When creating migrations, use alembic revision --autogenerate -m "message". The autogenerate will detect model changes and generate the migration script. However, be cautious with async-specific features like async def in migration scripts—avoid them unless necessary.

from logging.config import fileConfig
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
import asyncio

config = context.config
fileConfig(config.config_file_name)

from app.models import Base  # your models

target_metadata = Base.metadata

DATABASE_URL = config.get_main_option("sqlalchemy.url")

def run_migrations_offline():
    context.configure(url=DATABASE_URL, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

async def run_migrations_online():
    connectable = create_async_engine(DATABASE_URL)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())

Connection Pooling and Performance Tuning

Async SQLAlchemy uses a connection pool managed by asyncpg. Proper configuration is critical for performance.

Example: ``python engine = create_async_engine( DATABASE_URL, pool_size=10, max_overflow=20, pool_timeout=60, pool_recycle=3600, pool_pre_ping=True ) ``

For high-concurrency apps, consider using a larger pool size. Monitor pool usage with: ``python engine.pool.status() # returns a dict with pool info ``

Query optimization: Use selectinload for relationships, avoid lazy loading in async contexts. Use limit and offset for pagination. Use execution_options to set statement timeout per query: ``python result = await session.execute(stmt.execution_options(statement_timeout=5000)) ``

Batch operations: Use insert().returning() for bulk inserts: ``python from sqlalchemy import insert stmt = insert(User).returning(User.id, User.name) result = await session.execute(stmt, [{"name": "Alice"}, {"name": "Bob"}]) ``

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=10,
    max_overflow=20,
    pool_timeout=60,
    pool_recycle=3600,
    pool_pre_ping=True,
    echo=False
)

# Check pool status
print(engine.pool.status())

# Example of batch insert with returning
from sqlalchemy import insert

async def bulk_create_users(names: list):
    async with async_session() as session:
        stmt = insert(User).returning(User.id, User.name)
        result = await session.execute(stmt, [{"name": name} for name in names])
        await session.commit()
        return result.fetchall()

Error Handling and Retry Logic

In production, transient errors like connection timeouts, deadlocks, and serialization failures are common. Implement retry logic with exponential backoff.

Use tenacity library for robust retries:

```python from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from sqlalchemy.exc import OperationalError, DeadlockError

@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type((OperationalError, DeadlockError)) ) async def execute_with_retry(session, stmt): return await session.execute(stmt) ```

For transaction retries, you need to restart the entire transaction:

```python from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) async def transfer_with_retry(sender_id, receiver_id, amount): async with async_session() as session: async with session.begin(): sender = await session.get(User, sender_id) receiver = await session.get(User, receiver_id) sender.balance -= amount receiver.balance += amount ```

Note: The retry decorator will re-execute the entire function on failure, creating a new session.

Always log exceptions for debugging.

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from sqlalchemy.exc import OperationalError, DeadlockError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((OperationalError, DeadlockError))
)
async def safe_transfer(sender_id: int, receiver_id: int, amount: float):
    async with async_session() as session:
        async with session.begin():
            sender = await session.get(User, sender_id)
            receiver = await session.get(User, receiver_id)
            sender.balance -= amount
            receiver.balance += amount

# Usage
await safe_transfer(1, 2, 100.0)

Testing Async Database Code

Testing async database code requires async test runners like pytest-asyncio. Install: pip install pytest-asyncio.

Use fixtures to create a test database and session:

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

@pytest.fixture async def engine(): engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/testdb") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield engine async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await engine.dispose()

@pytest.fixture async def session(engine): async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with async_session() as session: yield session

@pytest.mark.asyncio async def test_create_user(session): user = User(name="Test") session.add(user) await session.commit() result = await session.execute(select(User).where(User.name == "Test")) assert result.scalar_one().name == "Test" ```

Use pytest.mark.asyncio decorator for async tests. For mocking, use unittest.mock with async mocks.

Isolation: Use separate test database or transactions that roll back after each test. The fixture above creates tables before and drops after.

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

@pytest.fixture
async def engine():
    engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/testdb")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest.fixture
async def session(engine):
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session

@pytest.mark.asyncio
async def test_create_user(session):
    user = User(name="Alice")
    session.add(user)
    await session.commit()
    result = await session.execute(select(User).where(User.name == "Alice"))
    assert result.scalar_one().name == "Alice"

Production Patterns and Best Practices

Deploying async SQLAlchemy in production requires attention to several patterns:

1. Session per request: In web frameworks, create a session per request and close it after the response. Use dependency injection (e.g., FastAPI's Depends).

2. Use connection pooling wisely: Set pool size based on your database's max connections and application concurrency. Monitor with SELECT * FROM pg_stat_activity.

3. Graceful shutdown: Dispose the engine on app shutdown to close all connections: ``python async def shutdown(): await engine.dispose() ``

4. Health checks: Implement a health check endpoint that runs a simple query: ``python async def health_check(): async with async_session() as session: await session.execute(select(1)) return {"status": "ok"} ``

5. Use environment variables for database URLs, never hardcode.

6. Logging: Enable SQLAlchemy logging in development, but disable in production to avoid performance overhead. Use structured logging.

7. Avoid global sessions: Never share sessions across requests or tasks. Each coroutine should have its own session.

8. Use asyncio.gather for concurrent queries: When you need multiple independent queries, run them concurrently: ``python async def get_dashboard_data(user_id): async with async_session() as session: user_task = session.get(User, user_id) posts_task = session.execute(select(Post).where(Post.user_id == user_id)) user, posts = await asyncio.gather(user_task, posts_task) return user, posts.scalars().all() ``

import asyncio
from sqlalchemy import select

async def get_dashboard_data(user_id: int):
    async with async_session() as session:
        user_task = session.get(User, user_id)
        posts_task = session.execute(select(Post).where(Post.user_id == user_id))
        user, posts = await asyncio.gather(user_task, posts_task)
    return user, posts.scalars().all()

# FastAPI dependency example
from fastapi import Depends, FastAPI

app = FastAPI()

async def get_session():
    async with async_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    return result.scalar_one()