Instruction file imported from zamax14/Copilot-Tools (.github/instructions/sqlalchemy.instructions.md). Copyright stays with the author.

SQLAlchemy Guidelines

Modelo Base

Usar un modelo base con campos comunes:

from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class TimestampMixin:
    """Mixin para campos de auditoría."""
    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())


class BaseModel(TimestampMixin, Base):
    """Modelo base con ID y timestamps."""
    __abstract__ = True
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

Reglas

• Usar el estilo Mapped (SQLAlchemy 2.0+), no el legacy Column()
• Siempre definir __tablename__ explícitamente
• Relaciones con Mapped[list["Child"]] y relationship(back_populates=...)
• Usar Enum de Python para campos con valores fijos
• Índices explícitos para columnas usadas en queries frecuentes
• No lógica de negocio dentro de los modelos

Relaciones

class User(BaseModel):
    __tablename__ = "users"

    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    orders: Mapped[list["Order"]] = relationship(back_populates="user", cascade="all, delete-orphan")


class Order(BaseModel):
    __tablename__ = "orders"

    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    total: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    user: Mapped["User"] = relationship(back_populates="orders")

Async Session

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

engine = create_async_engine(settings.DATABASE_URL, echo=settings.DEBUG)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Repository Pattern

class UserRepository:
    """Capa de acceso a datos para usuarios."""

    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get_by_id(self, user_id: int) -> User | None:
        return await self.session.get(User, user_id)

    async def get_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, user: User) -> User:
        self.session.add(user)
        await self.session.flush()
        return user