Creando un AutoCRUD en FastAPI + SQLAlchemy

Un AutoCRUD sería una de las operaciones más interesantes y más obvias que podríamos utilizar, ya que hemos visto que todo está modular, como hicimos antes, en un archivo aparte como manejar nuestros propios WebSockets con FastAPI, con inyección de dependencias y definición de clases directamente en las rutas. Por lo tanto, este sería el siguiente paso lógico.

Existe un paquete llamado FastAPI CRUD Router que promete justamente eso: crear automáticamente rutas para eliminar, actualizar, obtener, crear, etc. Una vez instalado, puedes configurarlo con clases Pydantic (dependencies), registrarlo y ya tendrías un AutoCRUD funcional.

El problema es que este paquete es bastante antiguo y funciona con Pydantic v1:

https://github.com/awtkns/fastapi-crudrouter

Lo cual hoy en día representa un inconveniente. Personalmente, esto es lo que menos me gusta de los microframeworks (y FastAPI puede considerarse como tal) traen lo mínimo necesario para funcionar y, a partir de ahí, tú debes orquestarlo todo (como vimos con arquitectura limpia). Algo similar ocurre en Flask.

Por eso, antes de instalar cualquier paquete que parezca resolver algo “automáticamente”, revisa su repositorio en GitHub. Si ves que no tiene actualizaciones desde hace años, huye. En algún momento (si no es que ya) te va a fallar.

Entonces, ¿qué hacemos?

🛠️ Creación de un Autocrud Genérico

Dado que los paquetes externos están desactualizados, lo mejor es crear nuestro propio Autocrud. Es un proceso sencillo si entendemos cómo funcionan los genéricos en Python.

Creamos nuestro propio AutoCRUD, ya que realmente es bastante sencillo.

Voy a hacerlo todo en un solo archivo para simplificar. En un proyecto real, puedes estructurarlo como prefieras. El código completo lo tienes en el repositorio asociado al curso y libro completo que puedes buscar en esta publicación.

Definición de Modelos Base

Primero, creamos un BaseModel de Pydantic. Algo sencillo:

  • Modelo para Categoría
  • Modelo para Tarea

Ambos heredan de BaseModel, ya que simplemente modelan datos. No necesitamos herencia compleja ni lógica adicional.

mycrud.py

from pydantic import BaseModel, Field

class CategoryBase(BaseModel):
    name: str

class Category(CategoryBase):
    id: int = Field(..., ge=1) # Ensure id is greater than or equal to 1
    class Config:
        from_attributes = True

Uso de Genéricos para un CRUD Reutilizable

Aquí empieza lo interesante.

Vamos a crear una clase AutoCRUD que reciba:

  • El schema
  • El prefijo
  • Las etiquetas
  • Otros argumentos necesarios
# Definimos un tipo genérico para nuestros esquemas
T = TypeVar("T", bound=BaseModel)

class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.db: List[T] = []

Como trabajaremos con distintos modelos (Categoría, Tarea, etc.), necesitamos que nuestra clase sea genérica.

Para eso usamos TypeVar, donde:

  • El tipo base será BaseModel
  • El tipo concreto podrá ser cualquier modelo que herede de él

Esto nos permite crear un CRUD robusto, tipado y reutilizable.

Herencia Múltiple y Method Resolution Order (MRO)

Nuestra clase heredará de:

  • Generic
  • APIRouter

Aquí entra en juego el Method Resolution Order (MRO).

Cuando llamamos a super().__init__(), Python debe decidir qué constructor ejecutar primero. El orden depende de cómo declaramos la herencia.

En este caso:

class MyCRUDRouter(Generic[T], APIRouter):

Python buscará primero en Generic, luego en APIRouter.

Como Generic no tiene constructor relevante, terminará usando el de APIRouter, que es donde realmente nos interesan los parámetros como:

  • prefix
  • tags

Por eso los pasamos mediante *args y **kwargs.

¿Por qué las rutas están dentro del init?

Aquí viene la parte “rara”.

Las funciones como:

  • @self.post
  • @self.get
  • @self.put
  • @self.delete

están definidas dentro del __init__.

T = TypeVar("T", bound=BaseModel)

class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        ***
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        async def create(item: self.schema):
            ***
        @self.get("/", response_model=List[self.schema])
        async def get_all():
            ***
        @self.get("/{item_id}", response_model=self.schema)
        async def get_one(item_id: int):
            ***
        @self.put("/{item_id}", response_model=self.schema)
        async def update(item_id: int, updated_data: self.schema):
            ***

        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
           ***
    def _find_item(self, item_id: int) -> Optional[T]:
        ***

¿Por qué?

Porque los decoradores se ejecutan cuando Python lee la clase, no cuando creas la instancia. Si intentáramos usar self.schema o self.prefix fuera del constructor, en ese momento aún no existirían, y la aplicación fallaría.

Por eso, las rutas deben declararse dentro del __init__, donde ya tenemos acceso a:

  • self.schema
  • self.db
  • y demás configuraciones

Los decoradores de FastAPI (como @self.post) se ejecutan cuando Python lee la clase. Si los ponemos fuera del constructor, intentarían acceder al esquema (self.schema) antes de que este sea asignado al crear el objeto, provocando un error.

Si lo movemos fuera, simplemente explota todo.

Implementación Simulada (En Memoria)

Primero hacemos una versión simple usando una lista como base de datos en memoria:

  • create
  • get_all
  • get_one
  • update
  • delete

Todo trabaja usando el tipo genérico T.

Esto nos permite probar rápidamente el funcionamiento sin base de datos real.

Quedando el código como:

from pydantic import BaseModel, Field
from typing import List, Type, TypeVar, Generic, Optional
from fastapi import APIRouter, HTTPException, status, Depends

from typing import Generic, TypeVar, Type, List
from pydantic import BaseModel
from sqlalchemy.orm import Session
from sqlalchemy import select

T = TypeVar("T", bound=BaseModel)

class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.db: List[T] = []

        # --- CREATE ---
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        async def create(item: self.schema):
            self.db.append(item)
            return item

        # --- READ ALL ---
        @self.get("/", response_model=List[self.schema])
        async def get_all():
            return self.db

        # --- READ ONE ---
        @self.get("/{item_id}", response_model=self.schema)
        async def get_one(item_id: int):
            item = self._find_item(item_id)
            if not item:
                raise HTTPException(status_code=404, detail="Elemento no encontrado")
            return item

        # --- UPDATE ---
        @self.put("/{item_id}", response_model=self.schema)
        async def update(item_id: int, updated_data: self.schema):
            for index, item in enumerate(self.db):
                if getattr(item, 'id', None) == item_id:
                    # En Pydantic V2 usamos model_dump para actualizar
                    self.db[index] = updated_data
                    return updated_data
            raise HTTPException(status_code=404, detail="No se pudo actualizar: no encontrado")

        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
        async def delete(item_id: int):
            for index, item in enumerate(self.db):
                if getattr(item, 'id', None) == item_id:
                    self.db.pop(index)
                    return
            raise HTTPException(status_code=404, detail="No se pudo eliminar: no encontrado")

    def _find_item(self, item_id: int) -> Optional[T]:
        """Método auxiliar para buscar por ID."""
        return next((item for item in self.db if getattr(item, 'id', None) == item_id), None)

Versión con Base de Datos (SQLAlchemy)

El ejemplo anterior funciona en memoria, pero para un proyecto real debemos integrarlo con SQLAlchemy. Lo único que cambia realmente es la lógica dentro de las funciones de ruta:

  • Conversión: Transformamos el modelo de Pydantic a un modelo de SQLAlchemy.
  • Exclusión de IDs: Al crear (POST), debemos excluir el campo id para que la base de datos lo genere automáticamente como autoincremental.
  • Operaciones: Reemplazamos el manejo de listas por db.add(), db.commit() y db.refresh().

Luego sustituimos la lista en memoria por:

  • Modelo SQLAlchemy
  • Sesión de base de datos
  • Operaciones reales con commit

Aquí lo único que cambia es:

  • Convertir el modelo Pydantic a modelo SQLAlchemy
  • Guardarlo en la base de datos
  • Manejar correctamente el ID autoincremental

Tuvimos que excluir el id en el schema de creación, para que lo genere internamente la base de datos y no nosotros:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, relationship

# 1. Definimos la Clase Base (Recomendado en SQLAlchemy 2.0)
class Base(DeclarativeBase):
    pass

# 2. El Modelo de la Entidad
class CategoryModel(Base):
    __tablename__ = "categories"

    # Definimos las columnas
    id=Column(Integer, primary_key=True, index=True)
    name = Column(String(255), nullable=False, unique=True)

    # Opcional: Si quieres que una categoría tenga muchas tareas
    # tasks = relationship("TaskModel", back_populates="category")

    def __repr__(self):
        return f"<Category(id={self.id}, name='{self.name}')>"        
        
class SQLCRUDRouter(Generic[T, M], APIRouter):
    def __init__(self, schema: Type[T], model: Type[M], get_db_func, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.model = model
        self.get_db = get_db_func

        # --- CREATE ---
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        def create(item: self.schema, db: Session = Depends(self.get_db)):
            # Convertimos Pydantic a Modelo de SQLAlchemy
            
            # 2. Creamos la instancia del modelo de SQLAlchemy con los datos limpios
            # db_item = self.model(**item.model_dump()) # hay que quitar el id
            db_item = self.model(**item.model_dump(exclude={'id'}, exclude_unset=True))
    
            
          
            db.add(db_item)
            db.commit()
            db.refresh(db_item)
            return db_item

        # --- READ ALL ---
        @self.get("/", response_model=List[self.schema])
        def get_all(db: Session = Depends(self.get_db)):
            # Usamos la nueva sintaxis de SQLAlchemy 2.0 (select)
            result = db.execute(select(self.model)).scalars().all()
            return result

        # --- READ ONE ---
        @self.get("/{item_id}", response_model=self.schema)
        def get_one(item_id: int, db: Session = Depends(self.get_db)):
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="No encontrado")
            return db_item

        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
        def delete(item_id: int, db: Session = Depends(self.get_db)):
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="No encontrado")
            db.delete(db_item)
            db.commit()
            return
        # --- UPDATE ---
        @self.put("/{item_id}", response_model=self.schema)
        def update(item_id: int, updated_data: self.schema, db: Session = Depends(self.get_db)):
            # 1. Buscar el registro existente
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="No se pudo actualizar: no encontrado")

            # 2. Extraer los datos de Pydantic
            # Excluimos 'id' para evitar que intenten cambiar la PK de la fila
            # exclude_unset=True permite actualizaciones parciales si el esquema lo soporta
            update_dict = updated_data.model_dump(exclude={'id'}, exclude_unset=True)

            # 3. Actualizar los atributos del modelo dinámicamente
            for key, value in update_dict.items():
                setattr(db_item, key, value)

            # 4. Persistir
            db.commit()
            db.refresh(db_item)
            return db_item

Uso del AutoCRUD

Una vez creada la clase, simplemente hacemos:

# app.include_router(
#     MyCRUDRouter(schema=Category, prefix="/categories", tags=["Categories"])
# )
app.include_router(
    MyCRUDRouter(schema=Task, prefix="/tasks", tags=["Tasks"])
)
app.include_router(
    SQLCRUDRouter(schema=Category, model=CategoryModel,get_db_func=get_database_session, prefix="/categories", tags=["Categories"])
)

Y automáticamente tenemos:

  • POST
  • GET
  • GET by ID
  • PUT
  • DELETE

Totalmente funcional.

Conclusión

Crear un AutoCRUD en FastAPI es mucho más sencillo de lo que parece. No necesitamos paquetes desactualizados ni depender de terceros.

Con esto, obtendrás automáticamente:

  • Documentación Swagger: Rutas generadas para cada entidad con sus respectivos modelos.
  • Validación: Tipado fuerte gracias a los genéricos de Python.
  • Funcionalidad: Un CRUD completo (Crear, Leer, Actualizar, Eliminar) funcionando con apenas un par de líneas de código por entidad.

Con:

  • Genéricos
  • Herencia múltiple
  • APIRouter
  • Pydantic
  • SQLAlchemy

Podemos construir una solución robusta, tipada y reutilizable y lo más importante: entendiendo realmente cómo funciona por dentro.

Aprende a crear un AutoCRUD en FastAPI con genéricos, APIRouter, Pydantic y SQLAlchemy, sin depender de paquetes.

Acepto recibir anuncios de interes sobre este Blog.

Andrés Cruz

EN In english
Python FastApi CRUD Base de Datos

Su experiencia en este sitio será mejorada con el uso de cookies.