Integración con SQLAlchemy

En aplicaciones reales, necesitamos persistir datos en una base de datos. Litestar se integra con SQLAlchemy a través de Advanced Alchemy, una biblioteca que proporciona repositorios y patrones avanzados para trabajar con bases de datos de forma estructurada.

Instalación

Para usar SQLAlchemy con Litestar, instala el paquete con el extra sqlalchemy:

uv add "litestar[standard,sqlalchemy]"

Organización del código

Para proyectos de tamaño medio a grande, es recomendable organizar el código en módulos separados:

app/
├── __init__.py      # Aplicación Litestar
├── db.py            # Configuración de base de datos
├── models.py        # Modelos SQLAlchemy
├── repositories.py  # Repositorios y providers
├── dtos.py          # Data Transfer Objects
└── controllers.py   # Controladores y rutas

Esta estructura separa responsabilidades y facilita el mantenimiento a medida que la aplicación crece.

Configuración del plugin

Advanced Alchemy se integra con Litestar mediante un plugin que gestiona las sesiones de base de datos automáticamente.

# db.py
from advanced_alchemy.extensions.litestar import (
    SQLAlchemyPlugin,
    SQLAlchemySyncConfig,
)
 
sqlalchemy_config = SQLAlchemySyncConfig(
    connection_string="sqlite:///db.sqlite3",
    create_all=True,  # Crea las tablas automáticamente
)
 
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

# __init__.py
from litestar import Litestar
from app.db import sqlalchemy_plugin
 
app = Litestar(
    route_handlers=[],
    plugins=[sqlalchemy_plugin],
)

Conexiones de base de datos

SQLite: sqlite:///db.sqlite

PostgreSQL: postgresql+psycopg2://user:pass@localhost/dbname

MySQL: mysql+pymysql://user:pass@localhost/dbname

Modelos SQLAlchemy

Advanced Alchemy proporciona clases base con campos comunes como id, created_at y updated_at.

# models.py
from advanced_alchemy.base import BigIntAuditBase
from sqlalchemy.orm import Mapped, mapped_column
 
 
class User(BigIntAuditBase):
    """Modelo de usuario con campos de auditoría."""
 
    __tablename__ = "users"
 
    username: Mapped[str] = mapped_column(unique=True)
    fullname: Mapped[str]
    password: Mapped[str]
 
    # Heredados de BigIntAuditBase:
    # - id: Mapped[int] (autoincremental)
    # - created_at: Mapped[datetime]
    # - updated_at: Mapped[datetime]

Clases base disponibles:

BigIntAuditBase: ID int con campos de auditoría
BigIntBase: Solo ID int sin auditoría
UUIDAuditBase: ID UUID con auditoría
UUIDBase: Solo ID UUID

Campos de auditoría

Los campos created_at y updated_at se gestionan automáticamente por SQLAlchemy.

Repositorios

Los repositorios encapsulan las operaciones de base de datos para un modelo específico.

# repositories.py
from advanced_alchemy.repository import SQLAlchemySyncRepository
 
 
class UserRepository(SQLAlchemySyncRepository[User]):
    """Repositorio para operaciones de base de datos de usuarios."""
 
    model_type = User

El repositorio proporciona métodos para operaciones CRUD:

add(model) - Crear un registro
add_many(models) - Crear múltiples registros
get(id) - Obtener por ID
get_one(**filters) - Obtener uno con filtros
get_and_update(match_fields, **kwargs) - Obtener y actualizar en una operación
list(*filters) - Listar con filtros
list_and_count(*filters) - Listar y contar
update(model) - Actualizar registro
delete(id) - Eliminar por ID

Inyección de dependencias

Litestar inyecta automáticamente la sesión de base de datos. Creamos una función que provee el repositorio.

# repositories.py
from sqlalchemy.orm import Session
 
 
async def provide_user_repo(db_session: Session) -> UserRepository:
    """Provee el repositorio de usuarios con auto-commit activado."""
    return UserRepository(session=db_session, auto_commit=True)

Auto-commit

Usar auto_commit=True confirma automáticamente los cambios después de cada operación, eliminando la necesidad de llamar session.commit() manualmente.

Controlador con repositorio

Integramos el repositorio en un controlador para crear una API completa.

# controllers.py
from typing import Sequence
 
from litestar import Controller, delete, get, patch, post
from litestar.di import Provide
 
from app.models import User
from app.repositories import UserRepository, provide_user_repo
 
 
class UserController(Controller):
    """Controlador para operaciones de gestión de usuarios."""
 
    path = "/users"
    dependencies = {"user_repo": Provide(provide_user_repo)}
 
    @get("/")
    async def list_users(
        self,
        user_repo: UserRepository,
    ) -> Sequence[User]:
        """Lista todos los usuarios."""
        return user_repo.list()
 
    @get("/{user_id:int}")
    async def get_user(
        self,
        user_repo: UserRepository,
        user_id: int,
    ) -> User:
        """Obtiene un usuario por ID."""
        return user_repo.get(user_id)
 
    @post("/")
    async def create_user(
        self,
        user_repo: UserRepository,
        data: dict,
    ) -> User:
        """Crea un nuevo usuario."""
        return user_repo.add(User(**data))
 
    @patch("/{user_id:int}")
    async def update_user(
        self,
        user_repo: UserRepository,
        user_id: int,
        data: dict,
    ) -> User:
        """Actualiza un usuario."""
        user, _ = user_repo.get_and_update(
            match_fields="id",
            id=user_id,
            **data,
        )
        return user
 
    @delete("/{user_id:int}", status_code=204)
    async def delete_user(
        self,
        user_repo: UserRepository,
        user_id: int,
    ) -> None:
        """Elimina un usuario."""
        user_repo.delete(user_id)

get_and_update()

El método get_and_update() combina la obtención y actualización en una sola operación de base de datos, siendo más eficiente que llamar get() y update() por separado.

Integración con DTOs

Advanced Alchemy proporciona SQLAlchemyDTO para crear DTOs directamente desde modelos SQLAlchemy.

# dtos.py
from advanced_alchemy.extensions.litestar import SQLAlchemyDTO, SQLAlchemyDTOConfig
 
from app.models import User
 
 
class UserReadDTO(SQLAlchemyDTO[User]):
    """DTO para leer usuarios sin exponer la contraseña."""
 
    config = SQLAlchemyDTOConfig(exclude={"password"})
 
 
class UserCreateDTO(SQLAlchemyDTO[User]):
    """DTO para crear usuarios - excluye campos generados."""
 
    config = SQLAlchemyDTOConfig(
        exclude={"id", "created_at", "updated_at"},
    )
 
 
class UserUpdateDTO(SQLAlchemyDTO[User]):
    """DTO para actualizar usuarios - parcial y excluye inmutables."""
 
    config = SQLAlchemyDTOConfig(
        exclude={"id", "created_at", "password"},
        partial=True,
    )

# controllers.py
from litestar.dto import DTOData
 
from app.dtos import UserCreateDTO, UserReadDTO, UserUpdateDTO
 
 
class UserController(Controller):
    path = "/users"
    dependencies = {"user_repo": Provide(provide_user_repo)}
    return_dto = UserReadDTO  # Aplica a todas las respuestas
 
    @post("/", dto=UserCreateDTO)
    async def create_user(
        self,
        user_repo: UserRepository,
        data: DTOData[User],
    ) -> User:
        """Crea un nuevo usuario."""
        return user_repo.add(data.create_instance())
 
    @patch("/{user_id:int}", dto=UserUpdateDTO)
    async def update_user(
        self,
        user_repo: UserRepository,
        user_id: int,
        data: DTOData[User],
    ) -> User:
        """Actualiza un usuario."""
        user, _ = user_repo.get_and_update(
            match_fields="id",
            id=user_id,
            **data.as_builtins(),
        )
        return user

SQLAlchemyDTO vs DataclassDTO

SQLAlchemyDTO usa directamente tus modelos de SQLAlchemy, evitando duplicar definiciones. Usa SQLAlchemyDTOConfig para configurar qué campos excluir o incluir.

create_instance() vs as_builtins()

data.create_instance() crea directamente una instancia del modelo

data.as_builtins() retorna un diccionario con los datos validados

Manejo de excepciones

Advanced Alchemy lanza excepciones específicas que podemos capturar para devolver respuestas HTTP apropiadas.

from typing import Any
 
from advanced_alchemy.exceptions import DuplicateKeyError, NotFoundError
from litestar import Request, Response
 
 
def not_found_error_handler(
    _: Request[Any, Any, Any],
    __: NotFoundError,
) -> Response[Any]:
    """Maneja errores cuando un registro no existe."""
    return Response(
        status_code=404,
        content={"status_code": 404, "detail": "User not found"},
    )
 
 
def duplicate_error_handler(
    _: Request[Any, Any, Any],
    __: DuplicateKeyError,
) -> Response[Any]:
    """Maneja errores cuando se intenta crear un registro duplicado."""
    return Response(
        status_code=409,
        content={"status_code": 409, "detail": "User already exists"},
    )
 
 
class UserController(Controller):
    path = "/users"
    dependencies = {"user_repo": Provide(provide_user_repo)}
    return_dto = UserReadDTO
    exception_handlers = {
        NotFoundError: not_found_error_handler,
        DuplicateKeyError: duplicate_error_handler,
    }
    # ... endpoints

Excepciones comunes:

NotFoundError - No se encontró el registro
DuplicateKeyError - Violación de constraint único
ConflictError - Conflicto general en la operación

Exception handlers

Los exception handlers en el controlador capturan automáticamente las excepciones de Advanced Alchemy y las convierten en respuestas HTTP apropiadas.

Paginación

Advanced Alchemy soporta paginación con LimitOffset.

from advanced_alchemy.filters import LimitOffset
from litestar.pagination import OffsetPagination
from litestar.params import Parameter
 
 
def provide_limit_offset(
    current_page: int = Parameter(ge=1, default=1),
    page_size: int = Parameter(ge=1, le=100, default=10),
) -> LimitOffset:
    """Provee paginación offset/limit."""
    return LimitOffset(page_size, page_size * (current_page - 1))
 
 
class UserController(Controller):
    path = "/users"
    dependencies = {
        "user_repo": Provide(provide_user_repo),
        "limit_offset": Provide(provide_limit_offset),
    }
    return_dto = UserReadDTO
 
    @get("/")
    async def list_users(
        self,
        user_repo: UserRepository,
        limit_offset: LimitOffset,
    ) -> OffsetPagination[User]:
        """Lista usuarios con paginación."""
        results, total = user_repo.list_and_count(limit_offset)
        return OffsetPagination[User](
            items=results,
            total=total,
            limit=limit_offset.limit,
            offset=limit_offset.offset,
        )

Ejemplo de petición:

GET /users?current_page=2&page_size=20

Respuesta:

{
  "items": [...],
  "total": 150,
  "limit": 20,
  "offset": 20
}

Filtros avanzados

Los repositorios soportan filtros complejos para consultas específicas.

from advanced_alchemy.filters import BeforeAfter, SearchFilter
from datetime import datetime, timedelta
 
 
@get("/search")
async def search_users(
    self,
    user_repo: UserRepository,
    query: str,
) -> list[User]:
    """Busca usuarios por nombre."""
    filter_ = SearchFilter(field_name="fullname", value=query)
    results, _ = user_repo.list_and_count(filter_)
    return results
 
 
@get("/recent")
async def recent_users(
    self,
    user_repo: UserRepository,
) -> list[User]:
    """Usuarios creados en los últimos 7 días."""
    filter_ = BeforeAfter(
        field_name="created_at",
        after=datetime.now() - timedelta(days=7),
    )
    results, _ = user_repo.list_and_count(filter_)
    return results

Filtros disponibles:

SearchFilter - Búsqueda de texto
BeforeAfter - Filtros de fecha/hora
CollectionFilter - Filtrar por lista de valores
OrderBy - Ordenamiento

Lógica de negocio personalizada

Además de las operaciones CRUD básicas, puedes implementar endpoints con lógica de negocio específica.

from dataclasses import dataclass
 
from litestar.exceptions import HTTPException
 
 
@dataclass
class PasswordUpdate:
    """Estructura para actualizar contraseña."""
 
    current_password: str
    new_password: str
 
 
class UserController(Controller):
    # ... configuración del controlador
 
    @post("/{user_id:int}/update-password", status_code=204)
    async def update_password(
        self,
        user_id: int,
        data: PasswordUpdate,
        user_repo: UserRepository,
    ) -> None:
        """Actualiza la contraseña de un usuario."""
        user = user_repo.get(user_id)
 
        # Validar contraseña actual
        if user.password != data.current_password:
            raise HTTPException(
                detail="Invalid password",
                status_code=401,
            )
 
        # Actualizar contraseña
        user.password = data.new_password
        user_repo.update(user)

Este patrón permite combinar operaciones CRUD con validaciones y lógica de negocio específica de tu aplicación.

Dataclasses para requests personalizados

Usa @dataclass para definir estructuras de datos específicas para operaciones que no se mapean directamente a tus modelos.

Apuntes BD II

Explorador

Integración con SQLAlchemy

Instalación

Organización del código

Configuración del plugin

Modelos SQLAlchemy

Repositorios

Inyección de dependencias

Controlador con repositorio

Integración con DTOs

Manejo de excepciones

Paginación

Filtros avanzados

Lógica de negocio personalizada

Vista Gráfica

Tabla de Contenidos

Retroenlaces