Conceptos básicos de bases de datos

9 ejemplos de fundamentos de bases de datos en Python: 6 básicos y 3 intermedios.

Prerrequisitos

uv pip install "sqlalchemy>=2.0" alembic asyncpg psycopg[binary]

• Se asume el estilo SQLAlchemy 2.0 en toda esta sección.

Ejemplos básicos

1. Conexión DB-API

Conéctate con un controlador PEP 249.

import sqlite3
 
with sqlite3.connect("app.db") as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY, name TEXT)")
    conn.execute("INSERT INTO items(name) VALUES (?)", ("widget",))

• El gestor de contexto confirma al tener éxito.
• Las consultas parametrizadas evitan la inyección de SQL.
• SQLite solo para desarrollo local a escala.

Relacionado: SQLAlchemy Core - capa de motor

2. Transacciones

Confirma y revierte explícitamente.

conn.execute("BEGIN")
try:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

• Mantén las transacciones cortas.
• Una unidad lógica por transacción.
• Maneja excepciones con reversión.

Relacionado: Gestión de conexiones - alcance del pool

3. Motor SQLAlchemy

Crea un motor y ejecuta texto.

from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///app.db")
with engine.begin() as conn:
    conn.execute(text("SELECT 1"))

• El motor gestiona las conexiones.
• begin() confirma automáticamente.
• Usa cadenas de dialecto de URL de forma consistente.

4. Modelo Declarativo ORM

Mapea una clase a una tabla.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
class Base(DeclarativeBase):
    pass
 
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()

• Mapped[] anota las columnas.
• DeclarativeBase es SQLAlchemy 2.0.
• Crea tablas a través de metadatos o Alembic.

Relacionado: SQLAlchemy ORM 2.0 - sesiones

5. Consulta de Sesión

Lee filas con una sesión.

from sqlalchemy import select
from sqlalchemy.orm import Session
 
with Session(engine) as session:
    rows = session.scalars(select(Item)).all()

• La sesión rastrea el estado del objeto.
• select() reemplaza la API de consulta heredada.
• El gestor de contexto cierra la sesión.

6. Insertar Fila ORM

Persiste una nueva entidad.

with Session(engine) as session:
    session.add(Item(name="bolt"))
    session.commit()

• add() se adjunta a la sesión.
• commit() vacía el SQL.
• refresh() recarga las claves generadas.

Ejemplos intermedios

7. Inicialización de Alembic

Inicia el historial de migraciones.

# alembic init alembic
# edita alembic.ini sqlalchemy.url
# alembic revision --autogenerate -m "init"
# alembic upgrade head

• Autogenera diferencias entre modelos y la base de datos.
• Revisa las migraciones manualmente.
• Actualiza en el pipeline de despliegue.

Relacionado: Migraciones con Alembic - flujos de trabajo

8. Motor Asíncrono

Acceso no bloqueante a PostgreSQL.

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

• Controlador asyncpg para Postgres.
• Empareja con rutas FastAPI asíncronas.
• pool_pre_ping evita sockets obsoletos.

Relacionado: Acceso asíncrono a bases de datos - sesiones

9. Caché Redis

Caché para lecturas frecuentes.

import redis
r = redis.Redis.from_url("redis://localhost:6379/0")
r.setex("item:1", 60, "cached-payload")

• Las claves TTL expiran automáticamente.
• Patrón "cache aside" en servicios.
• Considera la caché como opcional; la fuente de verdad sigue siendo la base de datos.

Relacionado: NoSQL y Caché - patrones

Versiones de la pila: Esta página se escribió para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ y uv 0.6+.