Cómo Conectar Python con PostgreSQL: Guía paso a paso con ejemplos

PostgreSQL es la base de datos de código abierto más avanzada del mundo y la que más crece en adopción en 2026. Es la opción por defecto en la mayoría de plataformas cloud modernas — Supabase, Neon, Railway, Render — y el motor predilecto para proyectos con Python en la industria.

En esta guía vas a aprender a conectar Python con PostgreSQL desde cero: instalación, conexión segura, CRUD completo, tipos de datos avanzados exclusivos de PostgreSQL (JSONB, ARRAY, UUID), transacciones, connection pool y la forma ORM con SQLAlchemy. Todo con código real y funcional.

Esta guía completa la trilogía de Python con bases de datos relacionales. Si ya leíste cómo conectar Python con MySQL o conectar Python con SQL Server, verás que muchos conceptos son similares. Pero PostgreSQL tiene características únicas que lo hacen especialmente potente con Python. Y si no tienes claros los fundamentos de PostgreSQL, te recomendamos leer primero nuestra guía sobre qué es PostgreSQL y en qué se diferencia de MySQL.

Las librerías disponibles para Python + PostgreSQL

En esta guía usamos psycopg2 como base (el más extendido), mostramos las diferencias con psycopg3 donde importan y cubrimos SQLAlchemy para el enfoque ORM.

Paso 1: Preparar el entorno

Instalar PostgreSQL

# Opción A: PostgreSQL con Docker (recomendada para desarrollo)
docker run -d \
  --name postgres-dev \
  -e POSTGRES_USER=usuario \
  -e POSTGRES_PASSWORD=clave123 \
  -e POSTGRES_DB=tienda \
  -p 5432:5432 \
  -v postgres_data:/var/lib/postgresql/data \
  postgres:16-alpine

# Opción B: Instalación en Ubuntu/Debian
sudo apt update && sudo apt install postgresql postgresql-contrib -y
sudo systemctl start postgresql
sudo systemctl enable postgresql

# Crear usuario y base de datos en Ubuntu
sudo -u postgres psql -c "CREATE USER usuario WITH PASSWORD 'clave123';"
sudo -u postgres psql -c "CREATE DATABASE tienda OWNER usuario;"

# Opción C: Instalación en macOS con Homebrew
brew install postgresql@16
brew services start postgresql@16

# Verificar que PostgreSQL está corriendo
pg_isready -h localhost -p 5432

Instalar psycopg2

# Opción A: psycopg2-binary (más fácil — incluye las dependencias)
# Ideal para desarrollo y aprendizaje
pip install psycopg2-binary

# Opción B: psycopg2 (compilado desde fuente — mejor para producción)
# Requiere libpq-dev instalado en el sistema
sudo apt install libpq-dev python3-dev -y  # Linux
pip install psycopg2

# Instalar psycopg3 (versión moderna)
pip install psycopg[binary]

# Extras necesarios
pip install python-dotenv

# Verificar instalación
python -c "import psycopg2; print('psycopg2 versión:', psycopg2.__version__)"

Crear la base de datos y tabla de ejemplo

-- Conectar a PostgreSQL:
-- psql -h localhost -U usuario -d tienda
-- o usando pgAdmin / DBeaver

-- Tabla de productos con tipos exclusivos de PostgreSQL
CREATE TABLE IF NOT EXISTS productos (
    id          SERIAL PRIMARY KEY,             -- equivalente a AUTO_INCREMENT
    nombre      VARCHAR(100) NOT NULL,
    precio      NUMERIC(10,2) NOT NULL CHECK (precio > 0),
    stock       INTEGER DEFAULT 0,
    categoria   VARCHAR(50),
    etiquetas   TEXT[],                         -- ARRAY nativo de PostgreSQL
    metadatos   JSONB,                          -- JSONB indexable de PostgreSQL
    activo      BOOLEAN DEFAULT TRUE,
    creado_en   TIMESTAMPTZ DEFAULT NOW(),      -- con zona horaria
    uuid_prod   UUID DEFAULT gen_random_uuid()  -- UUID nativo
);

-- Índices útiles
CREATE INDEX idx_productos_categoria ON productos(categoria);
CREATE INDEX idx_productos_etiquetas ON productos USING GIN(etiquetas);
CREATE INDEX idx_productos_metadatos ON productos USING GIN(metadatos);

-- Datos de prueba
INSERT INTO productos (nombre, precio, stock, categoria, etiquetas, metadatos)
VALUES
('Laptop Pro 15',     899.99, 20,  'tecnologia',
 ARRAY['laptop', 'premium', 'oferta'],
 '{"ram": "16GB", "almacenamiento": "512GB", "procesador": "Intel i7"}'::jsonb),
('Mouse Inalámbrico',  29.99, 150, 'periféricos',
 ARRAY['mouse', 'wireless'],
 '{"dpi": 1600, "botones": 6, "color": "negro"}'::jsonb),
('Monitor 4K',        399.99, 25,  'tecnologia',
 ARRAY['monitor', '4k', 'premium'],
 '{"pulgadas": 27, "hz": 144, "panel": "IPS"}'::jsonb);

Paso 2: Establecer la conexión

import psycopg2
from psycopg2 import OperationalError
import os
from dotenv import load_dotenv

load_dotenv()

# ── Variables de entorno (.env) ──────────────────────────────
# DB_HOST=localhost
# DB_PORT=5432
# DB_NAME=tienda
# DB_USER=usuario
# DB_PASSWORD=clave123

def obtener_conexion():
    """Establece una conexión a PostgreSQL usando variables de entorno."""
    try:
        conn = psycopg2.connect(
            host     = os.getenv("DB_HOST", "localhost"),
            port     = int(os.getenv("DB_PORT", 5432)),
            dbname   = os.getenv("DB_NAME"),     # ← dbname (no database) en psycopg2
            user     = os.getenv("DB_USER"),
            password = os.getenv("DB_PASSWORD"),
            connect_timeout = 10                 # timeout de conexión en segundos
        )
        return conn

    except OperationalError as e:
        print(f"❌ Error de conexión: {e}")
        return None

# ── También puedes usar una connection string (DSN) ──────────
def obtener_conexion_dsn():
    """Conexión usando DSN string — útil para Supabase y plataformas cloud."""
    dsn = (
        f"host={os.getenv('DB_HOST')} "
        f"port={os.getenv('DB_PORT', 5432)} "
        f"dbname={os.getenv('DB_NAME')} "
        f"user={os.getenv('DB_USER')} "
        f"password={os.getenv('DB_PASSWORD')}"
    )
    return psycopg2.connect(dsn)

# ── Probar la conexión ───────────────────────────────────────
conn = obtener_conexion()
if conn:
    cursor = conn.cursor()
    cursor.execute("SELECT version(), current_database(), current_user")
    version, db, usuario = cursor.fetchone()
    print(f"✅ Conectado a PostgreSQL")
    print(f"   Versión:  {version[:30]}...")
    print(f"   Base de datos: {db}")
    print(f"   Usuario: {usuario}")
    cursor.close()
    conn.close()

Paso 3: El cursor y sus variantes

psycopg2 ofrece varios tipos de cursor. La elección impacta directamente en cómo accedes a los datos:

import psycopg2
import psycopg2.extras

conn = obtener_conexion()

# ── Cursor por defecto — devuelve tuplas ─────────────────────
cursor_tupla = conn.cursor()
cursor_tupla.execute("SELECT id, nombre, precio FROM productos LIMIT 1")
fila = cursor_tupla.fetchone()
print(fila[0])       # 1   ← acceso por índice
print(fila[1])       # "Laptop Pro 15"

# ── RealDictCursor — devuelve diccionarios reales ────────────
cursor_dict = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cursor_dict.execute("SELECT id, nombre, precio FROM productos LIMIT 1")
fila = cursor_dict.fetchone()
print(fila["id"])      # 1   ← acceso por nombre de columna
print(fila["nombre"])  # "Laptop Pro 15"
print(dict(fila))      # convertir a dict estándar

# ── NamedTupleCursor — devuelve namedtuples ──────────────────
cursor_named = conn.cursor(cursor_factory=psycopg2.extras.NamedTupleCursor)
cursor_named.execute("SELECT id, nombre, precio FROM productos LIMIT 1")
fila = cursor_named.fetchone()
print(fila.id)       # 1   ← acceso por atributo
print(fila.nombre)   # "Laptop Pro 15"

# ── DictCursor — diccionario con acceso por índice también ───
cursor_dict2 = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)

conn.close()

💡 Recomendación: usa RealDictCursor en la mayoría de los casos — es el más intuitivo y compatible con json.dumps() para APIs.

Paso 4: CRUD completo con psycopg2

📖 READ — leer datos

import psycopg2
import psycopg2.extras
import os
from dotenv import load_dotenv

load_dotenv()

def obtener_conexion():
    return psycopg2.connect(
        host=os.getenv("DB_HOST"), port=os.getenv("DB_PORT", 5432),
        dbname=os.getenv("DB_NAME"), user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD")
    )

# ── Obtener todos los productos ──────────────────────────────
def obtener_todos_los_productos():
    conn = obtener_conexion()
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    try:
        cursor.execute("""
            SELECT id, nombre, precio, stock, categoria, etiquetas
            FROM productos
            WHERE activo = TRUE
            ORDER BY nombre
        """)
        productos = cursor.fetchall()
        return [dict(p) for p in productos]  # convertir RealDictRow a dict

    except psycopg2.Error as e:
        print(f"❌ Error: {e}")
        return []

    finally:
        cursor.close()
        conn.close()

# ── Obtener por ID ───────────────────────────────────────────
def obtener_producto_por_id(producto_id):
    conn = obtener_conexion()
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    try:
        # placeholder en psycopg2 es %s (igual que MySQL, distinto a pyodbc)
        cursor.execute(
            "SELECT * FROM productos WHERE id = %s AND activo = TRUE",
            (producto_id,)
        )
        fila = cursor.fetchone()
        return dict(fila) if fila else None

    finally:
        cursor.close()
        conn.close()

# ── Búsqueda con filtros y paginación ───────────────────────
def buscar_productos(categoria=None, precio_max=None, pagina=1, por_pagina=10):
    conn = obtener_conexion()
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    try:
        query  = "SELECT id, nombre, precio, stock FROM productos WHERE activo = TRUE"
        params = []

        if categoria:
            query += " AND categoria = %s"
            params.append(categoria)

        if precio_max:
            query += " AND precio <= %s"
            params.append(precio_max)

        # Paginación con LIMIT y OFFSET — mismo que MySQL
        query += " ORDER BY nombre LIMIT %s OFFSET %s"
        params.extend([por_pagina, (pagina - 1) * por_pagina])

        cursor.execute(query, params)
        return [dict(p) for p in cursor.fetchall()]

    finally:
        cursor.close()
        conn.close()

# Probar
productos = obtener_todos_los_productos()
print(f"Total: {len(productos)} productos")
for p in productos:
    print(f"  [{p['id']}] {p['nombre']} — ${p['precio']}")

➕ CREATE — insertar datos

# ── Insertar un producto ─────────────────────────────────────
def insertar_producto(nombre, precio, stock, categoria, etiquetas=None, metadatos=None):
    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        # RETURNING id → devuelve el ID generado por SERIAL
        # Equivalente a OUTPUT INSERTED.id en SQL Server
        cursor.execute("""
            INSERT INTO productos (nombre, precio, stock, categoria, etiquetas, metadatos)
            VALUES (%s, %s, %s, %s, %s, %s)
            RETURNING id, creado_en
        """, (nombre, precio, stock, categoria,
              etiquetas,                                      # lista Python → TEXT[]
              psycopg2.extras.Json(metadatos) if metadatos else None))  # dict → JSONB

        resultado = cursor.fetchone()
        nuevo_id   = resultado[0]
        creado_en  = resultado[1]

        conn.commit()
        print(f"✅ Producto creado — ID: {nuevo_id}, Fecha: {creado_en}")
        return nuevo_id

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error al insertar: {e}")
        return None

    finally:
        cursor.close()
        conn.close()

# ── Insertar múltiples productos ─────────────────────────────
def insertar_varios_productos(lista):
    """
    lista: list de tuplas (nombre, precio, stock, categoria)
    """
    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        psycopg2.extras.execute_values(
            cursor,
            "INSERT INTO productos (nombre, precio, stock, categoria) VALUES %s",
            lista,
            template=None,
            page_size=100  # inserta en bloques de 100
        )
        conn.commit()
        print(f"✅ {cursor.rowcount} productos insertados")
        return cursor.rowcount

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error: {e}")
        return 0

    finally:
        cursor.close()
        conn.close()

# Probar
nuevo_id = insertar_producto(
    nombre    = "Auriculares BT",
    precio    = 59.99,
    stock     = 40,
    categoria = "audio",
    etiquetas = ["auriculares", "bluetooth", "oferta"],
    metadatos = {"bateria": "30h", "cancelacion_ruido": True, "color": "negro"}
)

nuevos = [
    ("SSD 1TB",    89.99, 35, "almacenamiento"),
    ("Cable HDMI",  9.99, 200, "cables"),
    ("Hub USB-C",  29.99, 75, "periféricos"),
]
insertar_varios_productos(nuevos)

✏️ UPDATE — actualizar datos

def actualizar_producto(producto_id, **campos):
    """Actualiza solo los campos especificados."""
    if not campos:
        return False

    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        sets   = ", ".join([f"{campo} = %s" for campo in campos.keys()])
        valores = list(campos.values()) + [producto_id]

        cursor.execute(
            f"UPDATE productos SET {sets} WHERE id = %s RETURNING id",
            valores
        )
        conn.commit()

        if cursor.fetchone():
            print(f"✅ Producto {producto_id} actualizado")
            return True

        print(f"⚠️ Producto {producto_id} no encontrado")
        return False

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error: {e}")
        return False

    finally:
        cursor.close()
        conn.close()

# ── Actualizar un campo JSONB (parcialmente) ─────────────────
def actualizar_metadato(producto_id, clave, valor):
    """Actualiza un campo dentro del JSONB sin reemplazarlo completo."""
    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        # || es el operador de fusión de JSONB en PostgreSQL
        cursor.execute("""
            UPDATE productos
            SET metadatos = metadatos || %s::jsonb
            WHERE id = %s
            RETURNING metadatos
        """, (psycopg2.extras.Json({clave: valor}), producto_id))

        conn.commit()
        nuevo_jsonb = cursor.fetchone()[0]
        print(f"✅ JSONB actualizado: {nuevo_jsonb}")
        return nuevo_jsonb

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error: {e}")
        return None

    finally:
        cursor.close()
        conn.close()

# ── Añadir elemento a un ARRAY ───────────────────────────────
def agregar_etiqueta(producto_id, nueva_etiqueta):
    """Añade un elemento al array de etiquetas si no existe."""
    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        cursor.execute("""
            UPDATE productos
            SET etiquetas = array_append(etiquetas, %s)
            WHERE id = %s AND NOT (%s = ANY(etiquetas))
            RETURNING etiquetas
        """, (nueva_etiqueta, producto_id, nueva_etiqueta))

        conn.commit()
        resultado = cursor.fetchone()
        if resultado:
            print(f"✅ Etiqueta añadida. Etiquetas: {resultado[0]}")

    finally:
        cursor.close()
        conn.close()

# Probar
actualizar_producto(1, precio=849.99, stock=18)
actualizar_metadato(1, "descuento", True)
agregar_etiqueta(1, "black-friday")

🗑️ DELETE — eliminar datos

def desactivar_producto(producto_id):
    """Borrado lógico — marca como inactivo sin eliminar el registro."""
    return actualizar_producto(producto_id, activo=False)

def eliminar_producto(producto_id):
    """Borrado físico — elimina el registro de la BD."""
    conn = obtener_conexion()
    cursor = conn.cursor()

    try:
        cursor.execute(
            "DELETE FROM productos WHERE id = %s RETURNING nombre",
            (producto_id,)
        )
        conn.commit()

        resultado = cursor.fetchone()
        if resultado:
            print(f"✅ Producto '{resultado[0]}' eliminado")
            return True

        print(f"⚠️ Producto {producto_id} no encontrado")
        return False

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error: {e}")
        return False

    finally:
        cursor.close()
        conn.close()

desactivar_producto(4)

Paso 5: Tipos de datos exclusivos de PostgreSQL

Aquí es donde Python + PostgreSQL brillan especialmente. Como ya vimos en nuestra guía de PostgreSQL vs MySQL, los tipos de datos avanzados de PostgreSQL no tienen equivalente en otras bases de datos.

JSONB — documentos JSON indexables

conn   = obtener_conexion()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

# Consultar dentro de JSONB con operadores nativos
cursor.execute("""
    SELECT nombre, precio,
           metadatos->>'ram' AS ram,
           metadatos->>'procesador' AS procesador,
           (metadatos->>'descuento')::boolean AS en_descuento
    FROM productos
    WHERE metadatos ? 'ram'              -- tiene la clave 'ram'
      AND (metadatos->>'ram') = '16GB'  -- valor exacto
""")
laptops_16gb = cursor.fetchall()

# Buscar dentro de JSONB con @> (contiene)
cursor.execute("""
    SELECT nombre, metadatos
    FROM productos
    WHERE metadatos @> '{"cancelacion_ruido": true}'::jsonb
""")
con_anc = cursor.fetchall()

# Actualizar valor anidado en JSONB con jsonb_set
cursor.execute("""
    UPDATE productos
    SET metadatos = jsonb_set(metadatos, '{stock_minimo}', '5', true)
    WHERE categoria = 'tecnologia'
    RETURNING nombre, metadatos->>'stock_minimo' AS stock_min
""")
conn.commit()
print("Actualizados:", cursor.fetchall())

conn.close()

ARRAY — listas nativas

conn   = obtener_conexion()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

# Buscar productos que contienen una etiqueta específica
cursor.execute("""
    SELECT nombre, etiquetas
    FROM productos
    WHERE 'premium' = ANY(etiquetas)     -- array contiene 'premium'
""")
premium = cursor.fetchall()

# Productos que tienen TODAS estas etiquetas
cursor.execute("""
    SELECT nombre, etiquetas
    FROM productos
    WHERE etiquetas @> ARRAY['laptop', 'premium']  -- contiene ambas
""")
laptops_premium = cursor.fetchall()

# Contar elementos del array
cursor.execute("""
    SELECT nombre, array_length(etiquetas, 1) AS num_etiquetas
    FROM productos
    WHERE etiquetas IS NOT NULL
    ORDER BY num_etiquetas DESC
""")
conteo = cursor.fetchall()

# Los arrays de PostgreSQL se convierten automáticamente a listas Python
for p in premium:
    print(f"{p['nombre']}: {p['etiquetas']}")  # etiquetas es una lista Python

conn.close()

UUID, rangos y búsqueda full-text

conn   = obtener_conexion()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

# UUID — ya generado automáticamente con gen_random_uuid()
cursor.execute("SELECT id, uuid_prod, nombre FROM productos LIMIT 3")
for p in cursor.fetchall():
    print(f"  [{p['id']}] UUID: {p['uuid_prod']} — {p['nombre']}")

# Búsqueda de texto completo (Full-Text Search) — solo PostgreSQL
cursor.execute("""
    SELECT nombre, precio,
           ts_rank(
               to_tsvector('spanish', nombre || ' ' || COALESCE(categoria, '')),
               to_tsquery('spanish', 'laptop | monitor')
           ) AS relevancia
    FROM productos
    WHERE to_tsvector('spanish', nombre || ' ' || COALESCE(categoria, ''))
          @@ to_tsquery('spanish', 'laptop | monitor')
    ORDER BY relevancia DESC
""")
resultados_fts = cursor.fetchall()
print(f"\nBúsqueda full-text: {len(resultados_fts)} resultados")

conn.close()

Paso 6: Transacciones con psycopg2

Las transacciones de PostgreSQL son las más robustas del ecosistema open source. Ya vimos el concepto en profundidad en el artículo sobre transacciones SQL y ACID. Aquí lo llevamos a código Python:

def procesar_venta(producto_id, cantidad, cliente_id):
    """Venta atómica: reduce stock y registra la transacción."""
    conn = obtener_conexion()
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

    try:
        # psycopg2 NO tiene autocommit por defecto — las transacciones son explícitas
        # No necesitas BEGIN — psycopg2 abre una transacción automáticamente

        # 1. Leer y bloquear la fila con FOR UPDATE
        cursor.execute("""
            SELECT stock, precio
            FROM productos
            WHERE id = %s AND activo = TRUE
            FOR UPDATE  -- bloquea la fila para esta transacción
        """, (producto_id,))

        producto = cursor.fetchone()

        if not producto:
            raise ValueError(f"Producto {producto_id} no existe")

        if producto["stock"] < cantidad:
            raise ValueError(
                f"Stock insuficiente. Disponible: {producto['stock']}, pedido: {cantidad}"
            )

        total = float(producto["precio"]) * cantidad

        # 2. Reducir el stock
        cursor.execute(
            "UPDATE productos SET stock = stock - %s WHERE id = %s",
            (cantidad, producto_id)
        )

        # 3. Registrar la venta con RETURNING
        cursor.execute("""
            INSERT INTO ventas (cliente_id, producto_id, cantidad, total)
            VALUES (%s, %s, %s, %s)
            RETURNING id, created_at
        """, (cliente_id, producto_id, cantidad, total))

        venta = cursor.fetchone()

        # 4. COMMIT solo si todo fue bien
        conn.commit()
        print(f"✅ Venta #{venta['id']} completada — Total: ${total:.2f}")
        return venta["id"]

    except ValueError as e:
        conn.rollback()
        print(f"❌ Error de negocio: {e}")
        return None

    except psycopg2.Error as e:
        conn.rollback()
        print(f"❌ Error de BD: {e.pgcode} — {e.pgerror}")
        return None

    finally:
        cursor.close()
        conn.close()

# ── Context manager — la forma más elegante ──────────────────
def procesar_venta_cm(producto_id, cantidad, cliente_id):
    """La misma función usando context manager de psycopg2."""
    with psycopg2.connect(...) as conn:
        with conn.cursor() as cursor:
            # Si no hay excepción → COMMIT automático al salir del with
            # Si hay excepción → ROLLBACK automático
            cursor.execute(...)
            conn.commit()

Paso 7: Connection Pool con psycopg2

from psycopg2 import pool

# Crear el pool al iniciar la aplicación (una sola vez)
connection_pool = pool.ThreadedConnectionPool(
    minconn = 2,    # conexiones mínimas siempre abiertas
    maxconn = 10,   # máximo de conexiones simultáneas
    host     = os.getenv("DB_HOST"),
    port     = os.getenv("DB_PORT", 5432),
    dbname   = os.getenv("DB_NAME"),
    user     = os.getenv("DB_USER"),
    password = os.getenv("DB_PASSWORD")
)

def obtener_del_pool():
    """Obtiene una conexión del pool."""
    return connection_pool.getconn()

def devolver_al_pool(conn):
    """Devuelve la conexión al pool para reutilizarla."""
    connection_pool.putconn(conn)

# Uso con context manager personalizado
from contextlib import contextmanager

@contextmanager
def conexion_pool():
    """Context manager para conexiones del pool."""
    conn = connection_pool.getconn()
    try:
        yield conn
    except Exception:
        conn.rollback()
        raise
    finally:
        connection_pool.putconn(conn)  # siempre devuelve al pool

# Uso limpio:
def obtener_productos():
    with conexion_pool() as conn:
        cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
        cursor.execute("SELECT id, nombre, precio FROM productos WHERE activo = TRUE")
        return [dict(p) for p in cursor.fetchall()]

Paso 8: psycopg3 — la versión moderna

# pip install psycopg[binary]

import psycopg

# Principales mejoras de psycopg3 vs psycopg2:
# 1. Soporte nativo async/await
# 2. Preparación de statements automática
# 3. Mejor manejo de tipos

# ── Conexión síncrona ────────────────────────────────────────
with psycopg.connect(
    host=os.getenv("DB_HOST"),
    dbname=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
) as conn:

    # psycopg3: row_factory para diccionarios
    from psycopg.rows import dict_row

    with conn.cursor(row_factory=dict_row) as cursor:
        cursor.execute("SELECT id, nombre, precio FROM productos WHERE activo = TRUE")
        productos = cursor.fetchall()
        for p in productos:
            print(f"[{p['id']}] {p['nombre']} — ${p['precio']}")

# ── Conexión asíncrona (para FastAPI, aiohttp) ───────────────
import asyncio
import psycopg
from psycopg.rows import dict_row

async def obtener_productos_async():
    async with await psycopg.AsyncConnection.connect(
        host=os.getenv("DB_HOST"),
        dbname=os.getenv("DB_NAME"),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD")
    ) as conn:
        async with conn.cursor(row_factory=dict_row) as cursor:
            await cursor.execute(
                "SELECT id, nombre, precio FROM productos WHERE categoria = %s",
                ("tecnologia",)
            )
            return await cursor.fetchall()

# asyncio.run(obtener_productos_async())

Paso 9: SQLAlchemy con PostgreSQL

# pip install sqlalchemy psycopg2-binary

from sqlalchemy import create_engine, Column, Integer, String, Numeric, Boolean
from sqlalchemy import text, ARRAY
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = (
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT', 5432)}/{os.getenv('DB_NAME')}"
)

engine  = create_engine(DATABASE_URL, echo=False, pool_size=10, max_overflow=20)
Base    = declarative_base()
Session = sessionmaker(bind=engine)

# ── Modelo con tipos nativos de PostgreSQL ───────────────────
class Producto(Base):
    __tablename__ = "productos"

    id        = Column(Integer, primary_key=True)
    nombre    = Column(String(100), nullable=False)
    precio    = Column(Numeric(10, 2), nullable=False)
    stock     = Column(Integer, default=0)
    categoria = Column(String(50))
    etiquetas = Column(ARRAY(String))    # ARRAY de PostgreSQL
    metadatos = Column(JSONB)            # JSONB de PostgreSQL
    activo    = Column(Boolean, default=True)
    uuid_prod = Column(UUID(as_uuid=True))

    def __repr__(self):
        return f""

# ── CRUD con SQLAlchemy y tipos PostgreSQL ───────────────────
def demo_sqlalchemy_postgres():
    session = Session()

    try:
        # CREATE con ARRAY y JSONB
        nuevo = Producto(
            nombre    = "Cámara Web 4K",
            precio    = 129.99,
            stock     = 15,
            categoria = "periféricos",
            etiquetas = ["camara", "4k", "streaming"],
            metadatos = {"fps": 60, "autofocus": True, "microfono": True}
        )
        session.add(nuevo)
        session.commit()
        print(f"✅ Creado: {nuevo}")

        # READ — filtrar por elemento dentro de ARRAY (requiere SQL nativo)
        streaming = session.execute(
            text("SELECT nombre, etiquetas FROM productos WHERE 'streaming' = ANY(etiquetas)")
        ).fetchall()
        print(f"\nProductos para streaming: {len(streaming)}")

        # READ — filtrar por JSONB
        con_autofocus = session.execute(
            text("SELECT nombre FROM productos WHERE metadatos @> '{\"autofocus\": true}'::jsonb")
        ).fetchall()
        print(f"Con autofocus: {len(con_autofocus)}")

        # UPDATE
        producto = session.query(Producto).filter(Producto.id == 1).first()
        if producto:
            producto.precio    = 849.99
            producto.etiquetas = (producto.etiquetas or []) + ["reacondicionado"]
            session.commit()

    except Exception as e:
        session.rollback()
        print(f"❌ Error: {e}")

    finally:
        session.close()

demo_sqlalchemy_postgres()

Errores comunes y cómo resolverlos

❌ psycopg2.OperationalError: could not connect to server

# PostgreSQL no está corriendo o la dirección es incorrecta
# Verificar:
pg_isready -h localhost -p 5432
# Si falla: sudo systemctl start postgresql

# En Docker: verificar que el contenedor está corriendo
docker ps | grep postgres
docker start postgres-dev  # si está parado

❌ psycopg2.errors.UndefinedTable: relation "productos" does not exist

# La tabla no existe en el schema actual
# Verificar:
cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'")
tablas = cursor.fetchall()
print(tablas)

# O verificar el schema de búsqueda:
cursor.execute("SHOW search_path")
print(cursor.fetchone())

❌ Olvidar conn.commit() — los cambios no se guardan

# psycopg2 NO tiene autocommit por defecto
# Toda transacción necesita commit() explícito

cursor.execute("INSERT INTO productos (nombre, precio) VALUES (%s, %s)", ("Test", 9.99))
# ❌ Sin commit: si cierras la conexión, el INSERT se pierde
conn.commit()  # ✅ imprescindible para escrituras

❌ Error al serializar JSONB con diccionarios Python

# ❌ Pasar el dict directamente puede fallar según la versión
cursor.execute("INSERT INTO productos (metadatos) VALUES (%s)", ({"key": "value"},))

# ✅ Usar psycopg2.extras.Json para serialización segura
import psycopg2.extras
cursor.execute(
    "INSERT INTO productos (metadatos) VALUES (%s)",
    (psycopg2.extras.Json({"key": "value"}),)
)

❌ psycopg2-binary vs psycopg2 en producción

# psycopg2-binary: incluye libpq compilada — fácil pero puede dar problemas
# en contenedores y algunos sistemas en producción

# ✅ Para producción: instala las dependencias del sistema y usa psycopg2
sudo apt install libpq-dev python3-dev -y
pip install psycopg2  # compila con las librerías del sistema

Comparativa final: Python + MySQL vs SQL Server vs PostgreSQL

Resumen: lo que aprendiste hoy

• ✅ Las librerías principales son psycopg2 (estable y maduro) y psycopg3 (moderno con async)
• ✅ Usa psycopg2-binary para desarrollo y psycopg2 compilado para producción
• ✅ El parámetro de conexión es dbname (no database) en psycopg2
• ✅ El placeholder es %s para todos los tipos — igual que MySQL, distinto a pyodbc
• ✅ Usa RealDictCursor para acceder a columnas por nombre en lugar de por índice
• ✅ RETURNING id devuelve el ID generado — equivalente a lastrowid en MySQL
• ✅ execute_values() inserta múltiples filas eficientemente — mucho mejor que un bucle
• ✅ Los tipos JSONB y ARRAY de PostgreSQL son accesibles directamente como dict y list Python
• ✅ psycopg2 NO tiene autocommit — siempre necesitas commit() explícito
• ✅ El connection pool con ThreadedConnectionPool mejora el rendimiento en aplicaciones concurrentes
• ✅ psycopg3 añade soporte async nativo — ideal para FastAPI y aplicaciones modernas

🧪 ¿Tienes los fundamentos de Python y SQL bien sólidos?

Conectar Python con PostgreSQL requiere dominar funciones, manejo de excepciones y estructuras de datos en Python, más los comandos SQL esenciales. Comprueba dónde estás:

👉 Test: Python Básico 👉 Test: SQL Básico 

¿Usas PostgreSQL en tu trabajo o proyectos? ¿Cuál característica exclusiva de PostgreSQL te pareció más útil: JSONB, ARRAY o el full-text search nativo? Cuéntanos en los comentarios 👇 — respondemos todos. 🚀