Tutorial: FastAPI + SQLAlchemy + Alembic — Setup completo

FastAPI + SQLAlchemy 2.0 + Alembic — Setup Completo

Cómo configurar un proyecto FastAPI con SQLAlchemy async, Alembic para migraciones, y Pydantic v2 para schemas. Desde cero hasta tener un CRUD funcionando.

Instalación

pip install fastapi uvicorn sqlalchemy[asyncio] asyncpg alembic pydantic-settings

Estructura

project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── database.py
│   ├── models.py
│   └── schemas.py
├── alembic/
│   ├── versions/
│   └── env.py
├── alembic.ini
└── requirements.txt

database.py

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from app.config import get_settings

engine = create_async_engine(
    get_settings().database_url,
    echo=get_settings().debug,
    pool_size=20,
    max_overflow=10,
)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

models.py

from datetime import datetime
from sqlalchemy import String, Text, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    description: Mapped[str | None] = mapped_column(Text, default=None)
    price: Mapped[float]
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
    )

schemas.py

from pydantic import BaseModel, Field
from datetime import datetime

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: str | None = None
    price: float = Field(gt=0)

class ItemUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None
    price: float | None = Field(None, gt=0)
    is_active: bool | None = None

class ItemResponse(BaseModel):
    id: int
    name: str
    description: str | None
    price: float
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}

CRUD endpoint completo

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.models import Item
from app.schemas import ItemCreate, ItemUpdate, ItemResponse

app = FastAPI()

@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(data: ItemCreate, db: AsyncSession = Depends(get_db)):
    item = Item(**data.model_dump())
    db.add(item)
    await db.flush()
    await db.refresh(item)
    return item

@app.get("/items", response_model=list[ItemResponse])
async def list_items(
    skip: int = 0, limit: int = 20,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Item).where(Item.is_active).offset(skip).limit(limit)
    )
    return result.scalars().all()

@app.patch("/items/{item_id}", response_model=ItemResponse)
async def update_item(
    item_id: int, data: ItemUpdate,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(404, "Item no encontrado")

    for key, value in data.model_dump(exclude_unset=True).items():
        setattr(item, key, value)

    await db.flush()
    await db.refresh(item)
    return item

Alembic setup

# Inicializar Alembic (solo una vez)
alembic init alembic

# Editar alembic/env.py: importar Base.metadata
# Editar alembic.ini: sqlalchemy.url = postgresql://...

# Crear migración
alembic revision --autogenerate -m "create items table"

# Aplicar migraciones
alembic upgrade head

# Ver historial
alembic history

alembic/env.py (fragmento clave)

# Agregar al inicio de env.py:
from app.database import Base
from app.models import Item  # Importar todos los modelos
target_metadata = Base.metadata