Persistencia con SQLite en FastAPI (SQLAlchemy 2.x)

¿Qué es SQLAlchemy y por qué lo usamos?

SQLAlchemy es una librería de Python para trabajar con bases de datos relacionales.

  • Permite usar ORM (Object Relational Mapping): representamos tablas como clases de Python y filas como objetos.
  • Podemos crear, leer, actualizar y eliminar registros usando código Python en lugar de SQL puro.
  • Se puede usar en modo Core (SQL manual) o en modo ORM (objetos). En este curso usaremos el ORM.

👉 Objetivo en FastAPI

  • Integrar un CRUD real con base de datos.
  • Facilitar la escalabilidad: si cambiamos de SQLite a Postgres o MySQL, el código casi no cambia.

Sobre SQLite

SQLite es un motor de base de datos relacional ligero y embebido. Almacena los datos en un único archivo, no requiere servidor y resulta ideal para desarrollo, pruebas y aplicaciones pequeñas o embebidas. En producción puede servir para cargas moderadas; si se requiere mayor concurrencia, conviene migrar a motores como PostgreSQL o MySQL.

Si querés repasar conceptos de SQLite (tipos, sentencias SQL, índices y más), te recomendamos el curso de TutorialesProgramacionYa: SQLite Ya.

🔹 Instalación

Antes de comenzar, instalamos las dependencias necesarias:

pip install fastapi "uvicorn[standard]" sqlalchemy

Explicación:

  • fastapi: framework de la API.
  • uvicorn[standard]: servidor ASGI para ejecutar FastAPI.
  • sqlalchemy: ORM para manejar SQLite (o cualquier otra BD relacional).

📂 Estructura del proyecto

Vamos a organizar el proyecto en módulos separados:

.
 main.py         Archivo principal de la aplicación FastAPI
 database.py     Configuración de la base de datos y sesión
 models.py       Modelos ORM (tablas de la BD con SQLAlchemy)
 schemas.py      Modelos de validación (Pydantic)
 crud.py         Funciones para acceder a la BD (lógica CRUD)

database.py — Configuración de la BD y sesiones

Objetivo: centralizar la conexión con SQLite y la creación de sesiones.

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

DATABASE_URL = "sqlite:///./app.db"  # archivo SQLite local

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # requerido por SQLite
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

# Dependencia de FastAPI para inyectar la sesión
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Ventaja: si mañana cambiamos SQLite por Postgres o MySQL, solo hay que modificar este archivo.

models.py — Tablas de la base de datos (ORM)

Objetivo: definir la estructura de las tablas como clases.

from sqlalchemy import Column, Integer, String, Float
from database import Base

class Producto(Base):
    __tablename__ = "productos"

    codigo = Column(Integer, primary_key=True, index=True, autoincrement=True)  se genera automáticamente
    descripcion = Column(String(100), nullable=False)
    precio = Column(Float, nullable=False)

Ventaja: cada tabla es una clase — fácil de mantener y expandir (p. ej. agregar Cliente, Categoría, etc.).

schemas.py — Validación con Pydantic

Objetivo: validar datos de entrada/salida sin depender de la BD.

from pydantic import BaseModel, Field

class ProductoBase(BaseModel):
    descripcion: str = Field(min_length=3, max_length=100)
    precio: float = Field(gt=0)

# Para crear un producto
class ProductoCreate(ProductoBase):
    pass

class ProductoUpdate(ProductoBase):
    pass

class ProductoOut(ProductoBase):
    codigo: int
    class Config:
        from_attributes = True  # permite convertir objetos ORM a Pydantic

Ventaja: separa validación de datos de la lógica de negocio. Podemos tener diferentes modelos para crear, actualizar y responder datos.

crud.py — Lógica CRUD (operaciones en la BD)

Objetivo: tener todas las funciones que interactúan con la base de datos en un lugar separado.

from sqlalchemy.orm import Session
from models import Producto
from schemas import ProductoCreate, ProductoUpdate

def get_productos(db: Session):
    return db.query(Producto).all()

def get_producto(db: Session, codigo: int):
    return db.query(Producto).filter(Producto.codigo == codigo).first()

def create_producto(db: Session, data: ProductoCreate):
    prod = Producto(**data.model_dump())  # SQLite genera el código automáticamente
    db.add(prod)
    db.commit()
    db.refresh(prod)  # refresca para obtener el código asignado
    return prod

def update_producto(db: Session, codigo: int, data: ProductoUpdate):
    prod = get_producto(db, codigo)
    if prod:
        prod.descripcion = data.descripcion
        prod.precio = data.precio
        db.commit()
        db.refresh(prod)
    return prod

def delete_producto(db: Session, codigo: int):
    prod = get_producto(db, codigo)
    if prod:
        db.delete(prod)
        db.commit()
    return prod

Ventaja: la lógica de negocio está separada de las rutas — más ordenado y mantenible.

main.py — Definición de la API

Objetivo: manejar las rutas HTTP y delegar el trabajo a crud.py.

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from database import Base, engine, get_db
import crud
from schemas import ProductoCreate, ProductoUpdate, ProductoOut

app = FastAPI(title="FastAPI + SQLite", version="1.0.0")

# Crear las tablas automáticamente
Base.metadata.create_all(bind=engine)

@app.get("/productos", response_model=list[ProductoOut])
def listar_productos(db: Session = Depends(get_db)):
    return crud.get_productos(db)

@app.get("/productos/{codigo}", response_model=ProductoOut)
def obtener_producto(codigo: int, db: Session = Depends(get_db)):
    prod = crud.get_producto(db, codigo)
    if not prod:
        raise HTTPException(status_code=404, detail="Producto no encontrado")
    return prod

@app.post("/productos", response_model=ProductoOut, status_code=status.HTTP_201_CREATED)
def crear_producto(data: ProductoCreate, db: Session = Depends(get_db)):
    return crud.create_producto(db, data)

@app.put("/productos/{codigo}", response_model=ProductoOut)
def actualizar_producto(codigo: int, data: ProductoUpdate, db: Session = Depends(get_db)):
    prod = crud.update_producto(db, codigo, data)
    if not prod:
        raise HTTPException(status_code=404, detail="Producto no encontrado")
    return prod

@app.delete("/productos/{codigo}", status_code=status.HTTP_204_NO_CONTENT)
def eliminar_producto(codigo: int, db: Session = Depends(get_db)):
    prod = crud.delete_producto(db, codigo)
    if not prod:
        raise HTTPException(status_code=404, detail="Producto no encontrado")
    return

Ventaja: main.py queda enfocado en las rutas HTTP — fácil de leer.

Una vez que ejecutemos la aplicación se crea la base de datos:

base de datos sqlite

🚀 Resumen

Archivo        Objetivo                                      Ventaja
database.py   Configuración de BD y sesiones                 Cambiar de motor sin tocar rutas
models.py     Tablas con SQLAlchemy ORM                      Definición clara de la BD
schemas.py    Validación con Pydantic                        API robusta y desacoplada de la BD
crud.py       Funciones CRUD                                 Reuso de lógica y limpieza en rutas
main.py       Rutas FastAPI                                  Solo HTTP, delega lo demás

Ejemplo de uso

Con este cambio, el campo codigo se genera automáticamente en SQLite. Al hacer un POST /productos, solo enviamos:

{
  "descripcion": "Lapicera azul",
  "precio": 1200
}

Y la API responde:

{
  "codigo": 1,
  "descripcion": "Lapicera azul",
  "precio": 1200
}

Consultando la documentación automática http://127.0.0.1:8000/docs

documentacion docs
Retornar