Trabajar con Bases de Datos en Python: SQLite, PostgreSQL y SQLAlchemy
Python ofrece acceso a bases de datos relacionales a través de la DB-API 2.0 (PEP 249), un estándar común para todos los adaptadores. SQLAlchemy añade una capa de abstracción que permite escribir SQL portátil o usar el ORM (tratado en la guía de SQLAlchemy ORM).
sqlite3 — base de datos integrada
import sqlite3
import pathlib
# Conectar (crea el archivo si no existe)
conn = sqlite3.connect('app.db')
# Habilitar WAL para mejor concurrencia en lecturas
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
# Usar context manager — commit automático al salir, rollback si hay excepción
with conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS usuarios (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nombre TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
creado TEXT DEFAULT (datetime('now'))
)
""")
# Insertar con parámetros (previene SQL injection)
with conn:
conn.execute("INSERT INTO usuarios (nombre, email) VALUES (?, ?)",
('Ana García', 'ana@ejemplo.com'))
# Insertar múltiples filas eficientemente
usuarios = [
('Carlos López', 'carlos@ejemplo.com'),
('Elena Martín', 'elena@ejemplo.com'),
]
conn.executemany("INSERT OR IGNORE INTO usuarios (nombre, email) VALUES (?, ?)", usuarios)
# Consultas
cur = conn.execute("SELECT id, nombre, email FROM usuarios ORDER BY nombre")
for fila in cur:
print(f"{fila[0]}: {fila[1]} <{fila[2]}>")
# Convertir a dict con row_factory
conn.row_factory = sqlite3.Row
cur = conn.execute("SELECT * FROM usuarios WHERE nombre LIKE ?", ('%García%',))
usuarios_dict = [dict(row) for row in cur]
conn.close()
Context manager personalizado para sqlite3
from contextlib import contextmanager
import sqlite3
@contextmanager
def get_db(ruta: str = 'app.db'):
conn = sqlite3.connect(ruta)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# Uso
with get_db() as db:
db.execute("INSERT INTO usuarios (nombre, email) VALUES (?, ?)",
('Test User', 'test@ejemplo.com'))
psycopg2 — PostgreSQL
pip install psycopg2-binary
import psycopg2
import psycopg2.extras # RealDictCursor, execute_values, etc.
DSN = "host=localhost port=5432 dbname=mi_bd user=postgres password=secreto"
# Conectar
conn = psycopg2.connect(DSN)
conn.autocommit = False # control manual de transacciones
try:
with conn.cursor() as cur:
# Crear tabla
cur.execute("""
CREATE TABLE IF NOT EXISTS archivos (
id SERIAL PRIMARY KEY,
nombre TEXT NOT NULL,
formato TEXT NOT NULL,
tamanyo BIGINT,
procesado BOOLEAN DEFAULT FALSE,
creado_en TIMESTAMPTZ DEFAULT NOW()
)
""")
conn.commit()
# Insertar con parámetros %s (psycopg2 estilo)
cur.execute(
"INSERT INTO archivos (nombre, formato, tamanyo) VALUES (%s, %s, %s) RETURNING id",
('foto.jpg', 'JPEG', 2_456_789)
)
nuevo_id = cur.fetchone()[0]
print(f"Archivo insertado con id={nuevo_id}")
# Insertar lote con execute_values (mucho más rápido que executemany)
datos = [('video.mp4', 'MP4', 104_857_600), ('doc.pdf', 'PDF', 512_000)]
psycopg2.extras.execute_values(
cur,
"INSERT INTO archivos (nombre, formato, tamanyo) VALUES %s",
datos,
page_size=100
)
conn.commit()
# Consultar con RealDictCursor (devuelve dict)
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as dcur:
dcur.execute("SELECT * FROM archivos WHERE formato=%s ORDER BY creado_en DESC", ('JPEG',))
for row in dcur:
print(dict(row))
except psycopg2.Error as e:
conn.rollback()
print(f"Error PostgreSQL: {e.pgcode} — {e.pgerror}")
raise
finally:
conn.close()
Connection pooling con psycopg2
from psycopg2 import pool
# Pool de 2 a 10 conexiones
connection_pool = pool.ThreadedConnectionPool(
minconn=2,
maxconn=10,
dsn=DSN
)
def ejecutar_consulta(sql: str, params: tuple = ()) -> list[dict]:
conn = connection_pool.getconn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(sql, params)
if cur.description:
return [dict(r) for r in cur.fetchall()]
conn.commit()
return []
except Exception:
conn.rollback()
raise
finally:
connection_pool.putconn(conn)
# Uso
filas = ejecutar_consulta("SELECT * FROM archivos WHERE formato=%s", ('JPEG',))
SQLAlchemy Core — SQL expresivo y portable
pip install sqlalchemy
from sqlalchemy import (
create_engine, MetaData, Table, Column,
Integer, String, BigInteger, Boolean, DateTime, Text,
select, insert, update, delete, and_, or_, func,
)
from sqlalchemy.dialects.postgresql import insert as pg_insert
from datetime import datetime, timezone
# Motor (SQLite local o PostgreSQL en producción)
engine = create_engine(
"sqlite:///app.db",
# "postgresql+psycopg2://user:pass@localhost/db",
echo=False, # True para loggear SQL en desarrollo
pool_size=5,
max_overflow=10,
pool_pre_ping=True, # verifica la conexión antes de usarla
)
metadata = MetaData()
# Definir tabla
archivos = Table('archivos', metadata,
Column('id', Integer, primary_key=True),
Column('nombre', String(255),nullable=False),
Column('formato', String(10), nullable=False),
Column('tamanyo', BigInteger),
Column('procesado', Boolean, default=False),
Column('creado_en', DateTime, default=datetime.now),
)
metadata.create_all(engine) # crea tablas si no existen
# Operaciones con context manager
with engine.connect() as conn:
# INSERT
conn.execute(insert(archivos).values(
nombre='imagen.png', formato='PNG', tamanyo=1_234_567
))
# INSERT múltiple
conn.execute(insert(archivos), [
{'nombre': 'audio.mp3', 'formato': 'MP3', 'tamanyo': 5_000_000},
{'nombre': 'video.webm', 'formato': 'WEBM', 'tamanyo': 25_000_000},
])
conn.commit()
# SELECT con filtros encadenados
stmt = (
select(archivos)
.where(and_(
archivos.c.formato.in_(['PNG', 'JPEG']),
archivos.c.tamanyo > 100_000,
))
.order_by(archivos.c.tamanyo.desc())
.limit(10)
)
rows = conn.execute(stmt).mappings().all()
for row in rows:
print(dict(row))
# UPDATE
conn.execute(
update(archivos)
.where(archivos.c.formato == 'PNG')
.values(procesado=True)
)
conn.commit()
# Agregaciones
stmt_agg = select(
archivos.c.formato,
func.count().label('total'),
func.sum(archivos.c.tamanyo).label('bytes_totales'),
).group_by(archivos.c.formato).order_by(func.count().desc())
for row in conn.execute(stmt_agg):
print(f"{row.formato}: {row.total} archivos, {row.bytes_totales/1024:.1f} KB")
Transacciones y savepoints
from sqlalchemy.exc import IntegrityError
with engine.begin() as conn: # begin() hace commit/rollback automático
try:
conn.execute(insert(archivos).values(nombre='a.jpg', formato='JPEG'))
# Savepoint — rollback parcial sin deshacer toda la transacción
savepoint = conn.begin_nested()
try:
conn.execute(insert(archivos).values(nombre='a.jpg', formato='JPEG')) # duplicado
savepoint.commit()
except IntegrityError:
savepoint.rollback() # solo deshace desde el savepoint
print("Duplicado ignorado — continuando transacción")
conn.execute(insert(archivos).values(nombre='b.jpg', formato='JPEG'))
# conn.commit() implícito al salir del with engine.begin()
except Exception:
# conn.rollback() implícito al salir con excepción
raise
Migraciones con Alembic
pip install alembic
alembic init migrations
# migrations/env.py — configurar target_metadata
from myapp.models import metadata
target_metadata = metadata
# Crear migración automáticamente (detecta cambios)
# alembic revision --autogenerate -m "add_procesado_column"
# Ejemplo de migración generada
# migrations/versions/abc123_add_procesado_column.py
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('archivos', sa.Column('comentario', sa.Text(), nullable=True))
op.create_index('ix_archivos_formato', 'archivos', ['formato'])
def downgrade():
op.drop_index('ix_archivos_formato', table_name='archivos')
op.drop_column('archivos', 'comentario')
# Aplicar migraciones
# alembic upgrade head
# alembic downgrade -1
# alembic history --verbose
Consultas parametrizadas — prevención de SQL injection
# ✅ CORRECTO — siempre parámetros enlazados
nombre = input("Nombre de usuario: ")
# sqlite3
cur.execute("SELECT * FROM usuarios WHERE nombre=?", (nombre,))
# psycopg2
cur.execute("SELECT * FROM usuarios WHERE nombre=%s", (nombre,))
# SQLAlchemy (cualquier base de datos)
conn.execute(select(usuarios).where(usuarios.c.nombre == nombre))
# ❌ NUNCA formatear SQL manualmente
# cur.execute(f"SELECT * FROM usuarios WHERE nombre='{nombre}'") # Vulnerable!
Buenas prácticas
- Siempre usar parámetros enlazados (?, %s, :nombre) — nunca concatenar variables directamente en el SQL.
with engine.begin()en SQLAlchemy para transacciones automáticas — hace commit si no hay excepción, rollback si la hay.- Connection pooling es esencial en aplicaciones web — evita abrir y cerrar una conexión por cada petición.
PRAGMA foreign_keys=ONen SQLite — está desactivado por defecto y las claves foráneas no se validan sin esto.- Alembic para migraciones — nunca alteres el schema de producción manualmente; usa scripts versionados y revisables.
pool_pre_ping=Trueen SQLAlchemy evita errores "connection already closed" tras períodos de inactividad.
Conversiones relacionadas
Conversiones frecuentes del catálogo: