Database Integration — Asynchronous SQLAlchemy + Alembic

Q: Is there a difference between Mapped[str | None] and Mapped[Optional[str]] ?

They are functionally equivalent. str | None is the Python 3.10+ syntax, while Optional[str] is the typing -compatible notation. The former is recommended.

A database is like a warehouse—if there are too few doors (connections), the goods (queries) get backed up; if the layout (indexes) is disorganized, finding a single item (piece of data) requires searching through the entire warehouse.

1. What You'll Learn

SQLAlchemy 2.0 Asynchronous Engine: create_async_engine, AsyncSession
ORM Model Declarations: DeclarativeBase, Mapped, mapped_column (New 2.0 Style)
Alembic Asynchronous Migration: env.py Configure run_migrations_online Asynchronous Mode
Asynchronous Session Management: async with / yield Dependency Injection Patterns
Alice Scenario: Asynchronous Model Design for PriceTracker's products/prices/users Three Tables

2. Alice's True Story

(1) Pain Point: Database Synchronization Slows Down Asynchronous APIs

Alice uses synchronous SQLAlchemy to interact with the database, and each query blocks the event loop for 50-100 ms. When the number of concurrent requests to PriceTracker reached 500, the benefits of asynchronous FastAPI were completely offset by the synchronous database operations, and the P99 latency skyrocketed from the expected 50 ms to 2,000 ms. Charlie noted that the database connection pool had been exhausted, and new requests were queued up waiting for connections.

(2) Solution for Asynchronous SQLAlchemy

SQLAlchemy 2.0 provides native asynchronous support: create_async_engine + AsyncSession. Database queries no longer block the event loop, allowing FastAPI to fully leverage its asynchronous advantages.

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

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

(3) Revenue

After migrating to asynchronous SQLAlchemy, PriceTracker's P99 latency dropped from 2,000 ms to 80 ms, database connection utilization increased from 30% to 90%, and single-node QPS rose from 500 to over 3,000.

3. Asynchronous Engines and Sessions

(1) Database Architecture ER Diagram

erDiagram
    users ||--o{ products : creates
    products ||--o{ prices : has
    users {
        int id PK
        string email UK
        string hashed_password
        string role
        string subscription
        datetime created_at
    }
    products {
        int id PK
        string name
        string category
        float base_price
        string description
        int user_id FK
        datetime created_at
    }
    prices {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }

(1) ▶ Example: Asynchronous Engine and Connection Pool Configuration

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://pricetracker:secret@localhost:5432/pricetracker"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,          # Set True for SQL logging in dev
    pool_size=20,        # Persistent connections
    max_overflow=10,     # Extra connections when pool exhausted
    pool_timeout=30,     # Wait time for available connection
    pool_recycle=3600,   # Recycle connections after 1 hour
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Access objects after commit
)

Output:

TEXT

# Execution Successful

(2) Synchronous vs. Asynchronous SQLAlchemy: A Comparison

Dimension	Synchronous SQLAlchemy	Asynchronous SQLAlchemy 2.0
Engine	`create_engine`	`create_async_engine`
Session	`Session`	`AsyncSession`
Search	`session.execute(stmt)`	`await session.execute(stmt)`
Submit	`session.commit()`	`await session.commit()`
Driver	psycopg2	asyncpg
Blocked	Yes	No
Connection Pool	QueuePool	AsyncAdaptedQueuePool

4. ORM Model Declaration (New 2.0 Style)

(1) DeclarativeBase + Mapped + mapped_column

SQLAlchemy 2.0 replaces the old Column() declaration with Mapped[type] and mapped_column(), unifying type hints with ORM mappings.

(1) ▶ Example: PriceTracker Three-Table Model

PYTHON

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Float, Integer, DateTime, ForeignKey, Index
from sqlalchemy.orm import relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(50), default="user")
    subscription: Mapped[str] = mapped_column(String(50), default="free")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    products: Mapped[list["Product"]] = relationship(back_populates="owner")

class Product(Base):
    __tablename__ = "products"
    __table_args__ = (
        Index("ix_products_category", "category"),
        Index("ix_products_name", "name"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    category: Mapped[str] = mapped_column(String(100), nullable=False)
    base_price: Mapped[float] = mapped_column(Float, nullable=False)
    description: Mapped[str | None] = mapped_column(String(2000), nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    owner: Mapped["User"] = relationship(back_populates="products")
    prices: Mapped[list["Price"]] = relationship(back_populates="product")

class Price(Base):
    __tablename__ = "prices"
    __table_args__ = (
        Index("ix_prices_product_id", "product_id"),
        Index("ix_prices_recorded_at", "recorded_at"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    product_id: Mapped[int] = mapped_column(Integer, ForeignKey("products.id"))
    price: Mapped[float] = mapped_column(Float, nullable=False)
    currency: Mapped[str] = mapped_column(String(3), default="USD")
    source: Mapped[str] = mapped_column(String(100))
    recorded_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    product: Mapped["Product"] = relationship(back_populates="prices")

Output:

TEXT

# Execution Successful

(2) V1 Old Style vs. V2 New Style

Dimension	V1 Old Style	V2 New Style
Base	`declarative_base()`	`class Base(DeclarativeBase)`
Field Declaration	`Column(Integer, primary_key=True)`	`Mapped[int] = mapped_column(...)`
Type hint	None	`Mapped[type]` Full type
Optional Fields	`Column(String, nullable=True)`	`Mapped[str \| None]`
Relationship	`relationship()`	`Mapped[list["X"]] = relationship()`

5. Asynchronous Session Management and Dependency Injection

(1) The "yield" Dependency Pattern

(1) ▶ Example: DB Session Dependency

PYTHON

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Output:

TEXT

# Function defined successfully

(2) ▶ Example: Using Asynchronous Sessions in Endpoints

PYTHON

from fastapi import FastAPI, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
):
    stmt = select(Product).where(Product.id == product_id)
    result = await db.execute(stmt)
    product = result.scalar_one_or_none()
    if not product:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Product not found")
    return {
        "id": product.id,
        "name": product.name,
        "base_price": product.base_price,
    }

Output:

TEXT

# Function defined successfully

6. Alembic Asynchronous Migrations

(1) Migration Workflow

flowchart LR
    A[alembic revision --autogenerate -m desc] --> B[Edit Migration File]
    B --> C[alembic upgrade head]
    C --> D[Apply to Database]
    D --> E{Need Rollback?}
    E -->|Yes| F[alembic downgrade -1]
    E -->|No| G[Continue Development]

(1) ▶ Example: Initializing Alembic

BASH

# Install Alembic
uv add alembic

# Initialize Alembic with async template
cd pricetracker
alembic init -t async alembic

Output:

TEXT

# Command executed successfully

(2) ▶ Example: Configuring async mode in alembic/env.py

PYTHON

# alembic/env.py (key sections)
from sqlalchemy.ext.asyncio import create_async_engine
from app.models import Base  # Import your models
from app.core.config import settings

target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_async_engine(settings.database_url)

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

    with connectable.connect() as connection:
        asyncio.run(do_run_migrations(connection))

Output:

TEXT

# Function defined successfully

(3) ▶ Example: Generating and Applying Migration

BASH

# Auto-generate migration from model changes
alembic revision --autogenerate -m "add users products prices tables"

# Apply migration
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Check current version
alembic current

Output:

TEXT

# Command executed successfully

❓ FAQ

Q What is the difference between asyncpg and psycopg?

A asyncpg is a purely asynchronous PostgreSQL driver that offers better performance; psycopg3 supports asynchronous operations but is based on a C library. asyncpg is recommended for production use (postgresql+asyncpg://).

Q What does expire_on_commit=False mean?

A By default, ORM object properties expire after a commit (and accessing them again triggers a query). Setting this to False keeps the properties accessible, preventing lazy-loading issues.

Q Does Alembic's autogenerate feature detect all changes?

A No. It detects the addition or removal of tables and columns, as well as changes to indexes. It does not detect column name renaming, changes to constraint semantics, and other such changes; these require manually editing the migration files.

Q How should the connection pool size be set?

A Formula: pool_size = (CPU cores * 2) + number of active disks. PriceTracker uses pool_size=20 + max_overflow=10 to handle millions of queries.

Q Is there a difference between Mapped[str | None] and Mapped[Optional[str]]?

A They are functionally equivalent. str | None is the Python 3.10+ syntax, while Optional[str] is the typing-compatible notation. The former is recommended.

Q Can synchronous SQLAlchemy code be used in an asynchronous context?

A You cannot directly call synchronous SQLAlchemy methods within an async function. Wrap them using run_in_executor or use the asynchronous API exclusively.

📖 Summary

SQLAlchemy 2.0 Asynchronous Engine (create_async_engine + AsyncSession) features a non-blocking event loop that fully leverages the asynchronous advantages of FastAPI
Mapped[type] + mapped_column() is the new 2.0 style, which unifies type hints with ORM mappings
The yield dependency pattern manages the lifecycle of asynchronous sessions: automatic commit, rollback, and close
Alembic asynchronous migrations are initialized using the -t async template, and the env.py template configures the asynchronous engine
PriceTracker's three-table model (users/products/prices) includes indexing strategies and supports queries on millions of records

📝 Exercises

Basic Exercise (Difficulty ⭐): Configure create_async_engine to connect to PostgreSQL, create async_sessionmaker, and write the get_db yield dependency. Hint: create_async_engine(DATABASE_URL, pool_size=20)
Advanced Exercise (Difficulty ⭐⭐): Define the Product and Price models for PriceTracker (using the DeclarativeBase + Mapped style), including foreign key relationships and indexes. Hint: ForeignKey("products.id") + relationship()
Challenge (Difficulty: ⭐⭐⭐): Initialize Alembic in asynchronous mode, configure env.py, use autogenerate to generate migration files for three tables, and apply them to the database using upgrade head. Hint: alembic init -t async alembic + Modify env.py's target_metadata

---|

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us

404 Not Found

Database Integration — Asynchronous SQLAlchemy + Alembic

1. What You'll Learn

2. Alice's True Story

(1) Pain Point: Database Synchronization Slows Down Asynchronous APIs

(2) Solution for Asynchronous SQLAlchemy

(3) Revenue

3. Asynchronous Engines and Sessions

(1) Database Architecture ER Diagram

(1) ▶ Example: Asynchronous Engine and Connection Pool Configuration

(2) Synchronous vs. Asynchronous SQLAlchemy: A Comparison

4. ORM Model Declaration (New 2.0 Style)

(1) DeclarativeBase + Mapped + mapped_column

(1) ▶ Example: PriceTracker Three-Table Model

(2) V1 Old Style vs. V2 New Style

5. Asynchronous Session Management and Dependency Injection

(1) The "yield" Dependency Pattern

(1) ▶ Example: DB Session Dependency

(2) ▶ Example: Using Asynchronous Sessions in Endpoints

6. Alembic Asynchronous Migrations

(1) Migration Workflow

(1) ▶ Example: Initializing Alembic

(2) ▶ Example: Configuring async mode in alembic/env.py

(3) ▶ Example: Generating and Applying Migration

❓ FAQ

📖 Summary

📝 Exercises

🙏 帮我们做得更好