Integración completa: API, SQLAlchemy y Jinja2

Configuración de SQLAlchemy y modelos

La integración de SQLAlchemy con FastAPI permite gestionar bases de datos relacionales de manera eficiente y elegante. SQLAlchemy 2.0 introduce una sintaxis moderna que aprovecha las características de Python 3.13 y se integra perfectamente con el ecosistema asíncrono de FastAPI.

Instalación y configuración inicial

Para comenzar, necesitamos configurar la conexión a la base de datos y el motor de SQLAlchemy. En este ejemplo utilizaremos SQLite por su simplicidad, aunque la configuración es similar para otras bases de datos.

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

# Configuración del motor de base de datos
DATABASE_URL = "sqlite:///./app.db"

# Crear el motor de SQLAlchemy
engine = create_engine(
    DATABASE_URL, 
    connect_args={"check_same_thread": False}  # Necesario solo para SQLite
)

# Configurar la sesión
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

La clase DeclarativeBase es el punto de partida para definir nuestros modelos en SQLAlchemy 2.0. Esta clase base proporciona la funcionalidad necesaria para mapear clases Python a tablas de base de datos:

class Base(DeclarativeBase):
    pass

Definición de modelos de datos

Los modelos de SQLAlchemy representan las tablas de nuestra base de datos como clases Python. Cada atributo de la clase corresponde a una columna de la tabla.

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

Tipos de datos y restricciones

SQLAlchemy ofrece diversos tipos de columnas que se mapean a los tipos de datos de la base de datos. Las restricciones aseguran la integridad de los datos:

from sqlalchemy import Text, Float, ForeignKey
from sqlalchemy.orm import relationship

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False, index=True)
    description = Column(Text, nullable=True)
    price = Column(Float, nullable=False)
    stock = Column(Integer, default=0)
    is_available = Column(Boolean, default=True)
    
    # Clave foránea
    category_id = Column(Integer, ForeignKey("categories.id"))
    
    # Relación
    category = relationship("Category", back_populates="products")

Configuración de relaciones entre modelos

Las relaciones permiten conectar diferentes modelos y navegar entre ellos de manera intuitiva. SQLAlchemy 2.0 mantiene la sintaxis familiar para definir relaciones:

class Category(Base):
    __tablename__ = "categories"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)
    description = Column(Text, nullable=True)
    
    # Relación inversa
    products = relationship("Product", back_populates="category")

Inicialización de la base de datos

Para que nuestros modelos se reflejen en la base de datos, necesitamos crear las tablas. El método create_all() genera automáticamente las tablas basándose en nuestros modelos:

def create_database():
    """Crear todas las tablas en la base de datos"""
    Base.metadata.create_all(bind=engine)

def get_db():
    """Obtener una sesión de base de datos"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Configuración de dependencias para FastAPI

La función get_db() es una dependencia que proporciona una sesión de base de datos a nuestros endpoints. Esta función utiliza el patrón de inyección de dependencias de FastAPI:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI(title="Mi Aplicación con SQLAlchemy")

# Crear las tablas al iniciar la aplicación
create_database()

@app.get("/users/")
async def get_users(db: Session = Depends(get_db)):
    # Aquí utilizaremos la sesión db para consultar usuarios
    return {"message": "Endpoint configurado correctamente"}

Modelo completo de configuración

Un archivo de configuración completo integraría todos estos elementos de manera organizada:

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text, Float, ForeignKey
from sqlalchemy.orm import DeclarativeBase, sessionmaker, relationship
from datetime import datetime

# Configuración de base de datos
DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

# Modelos
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    full_name = Column(String(100), nullable=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

class Category(Base):
    __tablename__ = "categories"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)
    description = Column(Text, nullable=True)
    
    products = relationship("Product", back_populates="category")

class Product(Base):
    __tablename__ = "products"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False, index=True)
    description = Column(Text, nullable=True)
    price = Column(Float, nullable=False)
    stock = Column(Integer, default=0)
    category_id = Column(Integer, ForeignKey("categories.id"))
    
    category = relationship("Category", back_populates="products")

# Funciones utilitarias
def create_database():
    Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Esta configuración proporciona la base sólida necesaria para integrar SQLAlchemy con FastAPI. Los modelos definidos representan un esquema básico que puede expandirse según las necesidades de la aplicación, mientras que la configuración de sesiones asegura un manejo eficiente de las conexiones a la base de datos.

CRUD con base de datos y templates

Las operaciones CRUD representan el núcleo de cualquier aplicación web que gestiona datos. En esta sección integraremos SQLAlchemy con Jinja2 para crear una interfaz web completa que permita crear, leer, actualizar y eliminar registros de la base de datos.

Operaciones de lectura (Read) con templates

La operación más común es mostrar datos de la base de datos en nuestras plantillas HTML. SQLAlchemy 2.0 utiliza la sintaxis moderna con select() para consultar datos:

from fastapi import FastAPI, Depends, Request
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from sqlalchemy import select

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/users")
async def list_users(request: Request, db: Session = Depends(get_db)):
    # Consulta moderna de SQLAlchemy 2.0
    stmt = select(User).where(User.is_active == True)
    result = db.execute(stmt)
    users = result.scalars().all()
    
    return templates.TemplateResponse(
        "users/list.html",
        {"request": request, "users": users, "title": "Lista de Usuarios"}
    )

El template correspondiente mostraría los datos de manera estructurada y legible:

<!-- templates/users/list.html -->
<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
    <style>
        table { width: 100%; border-collapse: collapse; }
        th, td { padding: 10px; border: 1px solid #ddd; text-align: left; }
        th { background-color: #f2f2f2; }
    </style>
</head>
<body>
    <h1>{{ title }}</h1>
    <a href="/users/new">Agregar Usuario</a>
    
    <table>
        <thead>
            <tr>
                <th>ID</th>
                <th>Usuario</th>
                <th>Email</th>
                <th>Estado</th>
                <th>Acciones</th>
            </tr>
        </thead>
        <tbody>
            {% for user in users %}
            <tr>
                <td>{{ user.id }}</td>
                <td>{{ user.username }}</td>
                <td>{{ user.email }}</td>
                <td>{{ "Activo" if user.is_active else "Inactivo" }}</td>
                <td>
                    <a href="/users/{{ user.id }}/edit">Editar</a>
                    <a href="/users/{{ user.id }}/delete" onclick="return confirm('¿Estás seguro?')">Eliminar</a>
                </td>
            </tr>
            {% endfor %}
        </tbody>
    </table>
</body>
</html>

Operaciones de creación (Create) con formularios

Para crear nuevos registros, necesitamos combinar formularios HTML con endpoints que procesen los datos. Primero, el endpoint que muestra el formulario:

from fastapi import Form
from fastapi.responses import RedirectResponse

@app.get("/users/new")
async def show_user_form(request: Request):
    return templates.TemplateResponse(
        "users/form.html",
        {"request": request, "title": "Nuevo Usuario", "user": None}
    )

@app.post("/users/new")
async def create_user(
    request: Request,
    username: str = Form(...),
    email: str = Form(...),
    full_name: str = Form(""),
    db: Session = Depends(get_db)
):
    # Verificar si el usuario ya existe
    stmt = select(User).where(User.username == username)
    existing_user = db.execute(stmt).scalar_one_or_none()
    
    if existing_user:
        return templates.TemplateResponse(
            "users/form.html",
            {
                "request": request, 
                "title": "Nuevo Usuario",
                "error": "El nombre de usuario ya existe",
                "username": username,
                "email": email,
                "full_name": full_name
            }
        )
    
    # Crear el nuevo usuario
    new_user = User(
        username=username,
        email=email,
        full_name=full_name if full_name else None
    )
    
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    
    return RedirectResponse(url="/users", status_code=303)

El formulario HTML proporciona una interfaz intuitiva para introducir datos:

<!-- templates/users/form.html -->
<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
    <style>
        form { max-width: 500px; margin: 20px auto; }
        .form-group { margin-bottom: 15px; }
        label { display: block; margin-bottom: 5px; font-weight: bold; }
        input[type="text"], input[type="email"] { 
            width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 4px; 
        }
        .btn { padding: 10px 20px; background-color: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; }
        .btn:hover { background-color: #0056b3; }
        .error { color: red; margin-bottom: 15px; }
    </style>
</head>
<body>
    <h1>{{ title }}</h1>
    
    {% if error %}
        <div class="error">{{ error }}</div>
    {% endif %}
    
    <form method="post">
        <div class="form-group">
            <label for="username">Nombre de Usuario:</label>
            <input type="text" id="username" name="username" value="{{ username or '' }}" required>
        </div>
        
        <div class="form-group">
            <label for="email">Email:</label>
            <input type="email" id="email" name="email" value="{{ email or '' }}" required>
        </div>
        
        <div class="form-group">
            <label for="full_name">Nombre Completo:</label>
            <input type="text" id="full_name" name="full_name" value="{{ full_name or '' }}">
        </div>
        
        <button type="submit" class="btn">{{ "Actualizar" if user else "Crear" }} Usuario</button>
        <a href="/users">Cancelar</a>
    </form>
</body>
</html>

Operaciones de actualización (Update)

Las actualizaciones requieren cargar los datos existentes en el formulario y procesarlos cuando se envían. El endpoint de edición maneja ambos casos:

@app.get("/users/{user_id}/edit")
async def show_edit_form(user_id: int, request: Request, db: Session = Depends(get_db)):
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()
    
    if not user:
        return RedirectResponse(url="/users", status_code=303)
    
    return templates.TemplateResponse(
        "users/form.html",
        {"request": request, "title": "Editar Usuario", "user": user}
    )

@app.post("/users/{user_id}/edit")
async def update_user(
    user_id: int,
    request: Request,
    username: str = Form(...),
    email: str = Form(...),
    full_name: str = Form(""),
    db: Session = Depends(get_db)
):
    # Buscar el usuario a actualizar
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()
    
    if not user:
        return RedirectResponse(url="/users", status_code=303)
    
    # Verificar nombre de usuario único (excluyendo el actual)
    stmt = select(User).where(User.username == username, User.id != user_id)
    existing_user = db.execute(stmt).scalar_one_or_none()
    
    if existing_user:
        return templates.TemplateResponse(
            "users/form.html",
            {
                "request": request,
                "title": "Editar Usuario",
                "user": user,
                "error": "El nombre de usuario ya existe",
                "username": username,
                "email": email,
                "full_name": full_name
            }
        )
    
    # Actualizar los campos
    user.username = username
    user.email = email
    user.full_name = full_name if full_name else None
    
    db.commit()
    
    return RedirectResponse(url="/users", status_code=303)

Operaciones de eliminación (Delete)

La eliminación de registros debe implementarse con cuidado, incluyendo confirmaciones para evitar borrados accidentales:

@app.get("/users/{user_id}/delete")
async def delete_user(user_id: int, db: Session = Depends(get_db)):
    stmt = select(User).where(User.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()
    
    if user:
        db.delete(user)
        db.commit()
    
    return RedirectResponse(url="/users", status_code=303)

Operaciones con relaciones

Cuando trabajamos con modelos relacionados, las consultas se vuelven más complejas pero SQLAlchemy 2.0 las maneja elegantemente:

@app.get("/products")
async def list_products(request: Request, db: Session = Depends(get_db)):
    # Consulta con join para incluir la categoría
    stmt = select(Product).join(Category).where(Product.is_available == True)
    result = db.execute(stmt)
    products = result.scalars().all()
    
    return templates.TemplateResponse(
        "products/list.html",
        {"request": request, "products": products, "title": "Lista de Productos"}
    )

@app.get("/products/new")
async def show_product_form(request: Request, db: Session = Depends(get_db)):
    # Obtener categorías para el select
    stmt = select(Category).order_by(Category.name)
    categories = db.execute(stmt).scalars().all()
    
    return templates.TemplateResponse(
        "products/form.html",
        {"request": request, "categories": categories, "title": "Nuevo Producto"}
    )

El template para productos con selección de categorías:

<!-- templates/products/form.html -->
<div class="form-group">
    <label for="category_id">Categoría:</label>
    <select id="category_id" name="category_id" required>
        <option value="">Selecciona una categoría</option>
        {% for category in categories %}
        <option value="{{ category.id }}" 
                {{ "selected" if product and product.category_id == category.id else "" }}>
            {{ category.name }}
        </option>
        {% endfor %}
    </select>
</div>

Manejo de errores y validaciones

Un sistema CRUD robusto debe manejar errores de manera elegante y proporcionar retroalimentación clara al usuario:

from sqlalchemy.exc import IntegrityError

@app.post("/products/new")
async def create_product(
    request: Request,
    name: str = Form(...),
    price: float = Form(...),
    category_id: int = Form(...),
    db: Session = Depends(get_db)
):
    try:
        # Validar que la categoría existe
        stmt = select(Category).where(Category.id == category_id)
        category = db.execute(stmt).scalar_one_or_none()
        
        if not category:
            categories = db.execute(select(Category)).scalars().all()
            return templates.TemplateResponse(
                "products/form.html",
                {
                    "request": request,
                    "categories": categories,
                    "error": "La categoría seleccionada no existe",
                    "name": name,
                    "price": price
                }
            )
        
        new_product = Product(name=name, price=price, category_id=category_id)
        db.add(new_product)
        db.commit()
        
        return RedirectResponse(url="/products", status_code=303)
        
    except IntegrityError:
        db.rollback()
        categories = db.execute(select(Category)).scalars().all()
        return templates.TemplateResponse(
            "products/form.html",
            {
                "request": request,
                "categories": categories,
                "error": "Error de integridad en los datos",
                "name": name,
                "price": price
            }
        )

Esta integración completa de SQLAlchemy con Jinja2 proporciona una base sólida para aplicaciones web que requieren gestión de datos. Los patrones mostrados pueden extenderse y adaptarse para manejar modelos más complejos y requisitos específicos de cada aplicación.