Conexión de FastAPI con SQLAlchemy

Configuración de la conexión a la base de datos

Para conectar FastAPI con SQLAlchemy necesitamos establecer una configuración básica que permita a nuestra aplicación comunicarse con la base de datos. Este proceso implica definir la URL de conexión, crear el motor de base de datos y configurar los metadatos necesarios.

Instalación de dependencias

Antes de configurar la conexión, necesitamos instalar las librerías necesarias. SQLAlchemy requiere drivers específicos según el tipo de base de datos que utilicemos:

pip install sqlalchemy

Para SQLite, que usaremos en nuestros ejemplos, no necesitamos drivers adicionales ya que Python incluye soporte nativo. SQLite es ideal para desarrollo y aprendizaje porque no requiere instalación de servidor de base de datos.

Estructura del archivo de configuración

La configuración de la conexión se organiza típicamente en un archivo separado llamado database.py. Esta separación de responsabilidades mantiene el código organizado y facilita el mantenimiento:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# URL de conexión a la base de datos
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# Crear el motor de base de datos
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False}
)

# Configurar el creador de sesiones
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base para los modelos
Base = declarative_base()

Componentes de la configuración

El motor de base de datos (engine) es el componente central que gestiona las conexiones. Para SQLite, incluimos el parámetro connect_args={"check_same_thread": False} porque SQLite por defecto solo permite conexiones desde el hilo que creó la base de datos, pero FastAPI puede usar múltiples hilos.

La URL de conexión sigue el formato estándar de SQLAlchemy. Para SQLite, sqlite:///./test.db crea un archivo de base de datos llamado test.db en el directorio actual. Los tres barras (///) indican una ruta relativa.

# Ejemplos de URLs para diferentes bases de datos
SQLITE_URL = "sqlite:///./app.db"  # Archivo local
MYSQL_URL = "mysql://usuario:password@localhost/nombre_db"
POSTGRESQL_URL = "postgresql://usuario:password@localhost/nombre_db"

Configuración del SessionLocal

El SessionLocal es una fábrica de sesiones que utilizaremos para crear instancias de sesión de base de datos. Los parámetros autocommit=False y autoflush=False nos dan control manual sobre cuándo se confirman los cambios:

SessionLocal = sessionmaker(
    autocommit=False,  # Control manual de transacciones
    autoflush=False,   # Control manual de sincronización
    bind=engine        # Vincula al motor de base de datos
)

Base declarativa para modelos

La Base declarativa es la clase padre de la que heredarán todos nuestros modelos de SQLAlchemy. Esta base contiene los metadatos necesarios para que SQLAlchemy pueda crear y gestionar las tablas:

Base = declarative_base()

# Los modelos heredarán de esta base
class Usuario(Base):
    __tablename__ = "usuarios"
    # ... definición del modelo

Creación de tablas

Para que las tablas se creen en la base de datos, necesitamos llamar al método create_all() del motor. Esto se hace típicamente al iniciar la aplicación:

# Función para crear las tablas
def create_tables():
    Base.metadata.create_all(bind=engine)

Esta función lee los metadatos de todos los modelos que heredan de Base y crea las tablas correspondientes si no existen. Es una operación idempotente, lo que significa que es seguro ejecutarla múltiples veces.

Configuración completa del archivo database.py

El archivo completo de configuración queda estructurado de la siguiente manera:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Configuración de la base de datos
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# Motor de base de datos
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

# Fábrica de sesiones
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

# Base para modelos
Base = declarative_base()

# Función para crear tablas
def create_tables():
    Base.metadata.create_all(bind=engine)

Esta configuración establece los fundamentos necesarios para que FastAPI pueda interactuar con SQLAlchemy. El motor gestiona las conexiones, SessionLocal proporciona sesiones para las operaciones de base de datos, y Base sirve como punto de partida para definir modelos.

Creación de la sesión de base de datos

Una vez configurada la conexión, necesitamos crear sesiones de base de datos para realizar operaciones. Las sesiones son el mecanismo que SQLAlchemy utiliza para gestionar las transacciones y mantener el estado de los objetos durante las operaciones con la base de datos.

Función generadora de sesiones

Para integrar las sesiones con FastAPI, creamos una función generadora que proporciona una sesión y se encarga de cerrarla automáticamente. Esta función utiliza la palabra clave yield para crear un generador:

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

Esta función crea una nueva sesión usando SessionLocal(), la proporciona mediante yield, y garantiza que se cierre en el bloque finally sin importar si ocurre una excepción.

Integración con el sistema de dependencias

FastAPI utiliza su sistema de inyección de dependencias para gestionar las sesiones automáticamente. Importamos Depends y lo utilizamos en nuestros endpoints:

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

app = FastAPI()

@app.get("/usuarios/")
def obtener_usuarios(db: Session = Depends(get_db)):
    # La sesión 'db' está disponible para usar
    return {"mensaje": "Sesión de base de datos activa"}

El parámetro db: Session = Depends(get_db) indica a FastAPI que debe ejecutar la función get_db() y pasar el resultado como argumento db al endpoint.

Operaciones básicas con sesiones

Las sesiones proporcionan métodos para realizar consultas y modificar datos. Los métodos más comunes son query(), add(), commit() y rollback():

@app.post("/usuarios/")
def crear_usuario(nombre: str, db: Session = Depends(get_db)):
    # Crear un nuevo objeto (asumiendo que tenemos un modelo Usuario)
    nuevo_usuario = Usuario(nombre=nombre)
    
    # Agregar a la sesión
    db.add(nuevo_usuario)
    
    # Confirmar los cambios
    db.commit()
    
    # Refrescar para obtener el ID generado
    db.refresh(nuevo_usuario)
    
    return {"id": nuevo_usuario.id, "nombre": nuevo_usuario.nombre}

Gestión de transacciones

Las sesiones manejan transacciones automáticamente. Cuando llamamos commit(), todos los cambios pendientes se guardan en la base de datos. Si ocurre un error, podemos usar rollback() para deshacer los cambios:

@app.put("/usuarios/{usuario_id}")
def actualizar_usuario(usuario_id: int, nombre: str, db: Session = Depends(get_db)):
    try:
        # Buscar el usuario
        usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
        
        if not usuario:
            return {"error": "Usuario no encontrado"}
        
        # Modificar el usuario
        usuario.nombre = nombre
        
        # Confirmar cambios
        db.commit()
        
        return {"mensaje": "Usuario actualizado"}
    
    except Exception as e:
        # Deshacer cambios en caso de error
        db.rollback()
        return {"error": "Error al actualizar usuario"}

Consultas con sesiones

Para realizar consultas, utilizamos el método query() de la sesión. Este método nos permite filtrar, ordenar y limitar los resultados:

@app.get("/usuarios/{usuario_id}")
def obtener_usuario(usuario_id: int, db: Session = Depends(get_db)):
    # Consulta por ID específico
    usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
    
    if not usuario:
        return {"error": "Usuario no encontrado"}
    
    return {"id": usuario.id, "nombre": usuario.nombre}

@app.get("/usuarios/")
def listar_usuarios(db: Session = Depends(get_db)):
    # Consulta todos los usuarios
    usuarios = db.query(Usuario).all()
    
    return [{"id": u.id, "nombre": u.nombre} for u in usuarios]

Archivo completo de ejemplo

Integrando todo lo anterior, un archivo main.py completo quedaría así:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import get_db, create_tables

app = FastAPI()

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

@app.get("/")
def root():
    return {"mensaje": "API con SQLAlchemy funcionando"}

@app.get("/test-db/")
def test_database(db: Session = Depends(get_db)):
    # Verificar que la conexión funciona
    try:
        # Ejecutar una consulta simple
        result = db.execute("SELECT 1")
        return {"estado": "Conexión exitosa", "resultado": result.scalar()}
    except Exception as e:
        return {"error": f"Error de conexión: {str(e)}"}

Ciclo de vida de las sesiones

Es importante entender que cada request HTTP obtiene su propia sesión. FastAPI se encarga de:

• Crear una nueva sesión al inicio del request
• Proporcionar la sesión a todos los endpoints que la necesiten
• Cerrar automáticamente la sesión al finalizar el request

Esta gestión automática garantiza que no tengamos conexiones abiertas innecesariamente y que cada request opere de forma aislada.