Conectar Python con SQL Server permite automatizar consultas, importar datos, generar informes y construir aplicaciones que utilizan una base de datos empresarial. La biblioteca pyodbc actúa como puente entre Python y un controlador ODBC instalado en el sistema. Una integración confiable debe proteger credenciales, utilizar consultas parametrizadas, controlar transacciones y cerrar conexiones incluso cuando ocurre un error.
Antes de comenzar, revisa la guía para leer variables de entorno, el tutorial de manejo de excepciones, la introducción a Pandas, el artículo sobre funciones y la guía de logging.
Las referencias externas principales son la guía oficial de Microsoft para Python y pyodbc y la documentación del proyecto pyodbc.
Requisitos de la conexión
Necesitas Python, la biblioteca pyodbc, acceso al servidor y un controlador ODBC compatible instalado en el sistema operativo. Instalar el paquete de Python no instala automáticamente el controlador de SQL Server.
python -m pip install pyodbc
python -m pip show pyodbc
Consulta los controladores detectados:
import pyodbc
for controlador in pyodbc.drivers():
print(controlador)
El nombre utilizado en la cadena debe coincidir exactamente con uno de los disponibles.
No guardar credenciales en el código
Define variables de entorno:
SQLSERVER_HOST=servidor.ejemplo.local
SQLSERVER_DATABASE=ventas
SQLSERVER_USER=usuario_app
SQLSERVER_PASSWORD=secreto
SQLSERVER_DRIVER=ODBC Driver 18 for SQL Server
Mantén el archivo local fuera de Git y utiliza un gestor de secretos en producción.
Construir una conexión básica
import os
import pyodbc
def requerida(nombre):
valor = os.getenv(nombre)
if valor is None or not valor.strip():
raise RuntimeError(f"Falta la variable {nombre}")
return valor
def crear_conexion():
driver = requerida("SQLSERVER_DRIVER")
host = requerida("SQLSERVER_HOST")
database = requerida("SQLSERVER_DATABASE")
user = requerida("SQLSERVER_USER")
password = requerida("SQLSERVER_PASSWORD")
cadena = (
f"DRIVER={{{driver}}};"
f"SERVER={host};"
f"DATABASE={database};"
f"UID={user};"
f"PWD={password};"
"Encrypt=yes;"
"TrustServerCertificate=no;"
"Connection Timeout=10;"
)
return pyodbc.connect(cadena)
Los parámetros de cifrado dependen del entorno y del certificado configurado. No desactives verificaciones solo para eliminar un error; corrige la confianza del certificado o la configuración del servidor.
Utilizar autenticación integrada
En entornos Windows administrados, puede utilizarse autenticación integrada cuando el servidor y la identidad lo permiten:
cadena = (
f"DRIVER={{{driver}}};"
f"SERVER={host};"
f"DATABASE={database};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
)
La disponibilidad depende del sistema, controlador, dominio y configuración del servidor. La cuenta que ejecuta el proceso necesita permisos adecuados.
Ejecutar una consulta SELECT
def listar_clientes(limite=20):
consulta = """
SELECT TOP (?) id, nombre, email
FROM clientes
ORDER BY id DESC
"""
with crear_conexion() as conexion:
cursor = conexion.cursor()
cursor.execute(consulta, limite)
return cursor.fetchall()
Según el controlador y la sentencia, algunos elementos como TOP pueden requerir una estrategia distinta. Para filtros de valores, utiliza siempre parámetros.
Convertir filas en diccionarios
def filas_como_diccionarios(cursor):
columnas = [columna[0] for columna in cursor.description]
return [dict(zip(columnas, fila)) for fila in cursor.fetchall()]
Este formato facilita devolver JSON o trabajar con nombres de columnas. Para grandes resultados, itera por bloques en lugar de cargar todo en memoria.
Consultas parametrizadas
consulta = """
SELECT id, nombre, email
FROM clientes
WHERE ciudad = ? AND activo = ?
"""
cursor.execute(consulta, ciudad, True)
No construyas SQL concatenando strings:
# Inseguro: no usar
consulta = f"SELECT * FROM clientes WHERE ciudad = '{ciudad}'"
Los parámetros reducen el riesgo de inyección SQL y manejan correctamente comillas y tipos.
Insertar un registro
def crear_cliente(nombre, email):
consulta = """
INSERT INTO clientes (nombre, email)
OUTPUT INSERTED.id
VALUES (?, ?)
"""
with crear_conexion() as conexion:
cursor = conexion.cursor()
cursor.execute(consulta, nombre, email)
nuevo_id = cursor.fetchone()[0]
conexion.commit()
return nuevo_id
Valida longitud, formato y reglas antes de ejecutar. Una restricción en la base de datos sigue siendo necesaria para proteger la integridad.
Controlar transacciones
def transferir(origen, destino, importe):
conexion = crear_conexion()
try:
cursor = conexion.cursor()
cursor.execute(
"UPDATE cuentas SET saldo = saldo - ? WHERE id = ?",
importe,
origen,
)
cursor.execute(
"UPDATE cuentas SET saldo = saldo + ? WHERE id = ?",
importe,
destino,
)
conexion.commit()
except Exception:
conexion.rollback()
raise
finally:
conexion.close()
Las operaciones relacionadas deben confirmarse juntas. Si una falla, rollback() evita dejar datos a medias.
Actualizar y eliminar
cursor.execute(
"UPDATE clientes SET activo = ? WHERE id = ?",
False,
cliente_id,
)
cursor.execute(
"DELETE FROM sesiones WHERE expira < ?",
fecha_limite,
)
Antes de eliminar, comprueba el filtro y registra la cantidad afectada:
print("Filas afectadas:", cursor.rowcount)
Insertar múltiples filas
datos = [
("Ana", "[email protected]"),
("Luis", "[email protected]"),
]
with crear_conexion() as conexion:
cursor = conexion.cursor()
cursor.fast_executemany = True
cursor.executemany(
"INSERT INTO clientes (nombre, email) VALUES (?, ?)",
datos,
)
conexion.commit()
Prueba fast_executemany con tu controlador y volumen. Divide lotes enormes y captura qué registros fallaron.
Leer datos con Pandas
import pandas as pd
consulta = """
SELECT fecha, producto, cantidad, precio
FROM ventas
WHERE fecha >= ?
"""
with crear_conexion() as conexion:
df = pd.read_sql_query(
consulta,
conexion,
params=[fecha_inicio],
)
Selecciona solo columnas y filas necesarias. Para resultados grandes, utiliza chunksize.
Manejar errores específicos
import logging
import pyodbc
try:
with crear_conexion() as conexion:
conexion.execute("SELECT 1")
except pyodbc.InterfaceError:
logging.exception("Controlador o cadena de conexión inválidos")
except pyodbc.OperationalError:
logging.exception("Servidor no disponible o autenticación fallida")
except pyodbc.DatabaseError:
logging.exception("Error de base de datos")
No registres la cadena completa porque puede contener la contraseña.
Timeouts y bloqueos
Configura un tiempo de conexión y, cuando corresponda, un timeout para consultas. Las consultas lentas deberían analizarse con índices, planes de ejecución y filtros apropiados. Aumentar el timeout no resuelve un diseño ineficiente.
Pool de conexiones
pyodbc puede utilizar pooling ODBC. En aplicaciones web, administra conexiones por solicitud o mediante una capa adecuada. No compartas libremente un cursor entre threads y no mantengas transacciones abiertas mientras esperas entrada del usuario.
Permisos mínimos
La cuenta de la aplicación debería tener solo permisos necesarios. Separa cuentas de lectura, escritura y administración cuando el riesgo lo justifique. No utilices una cuenta administrativa para un informe de solo lectura.
Probar sin producción
Utiliza una base de pruebas con datos ficticios. Las pruebas de integración deben crear sus registros, verificar resultados y limpiar el estado. No ejecutes pruebas destructivas contra producción.
Errores frecuentes
Los fallos habituales son no instalar el controlador ODBC, escribir un nombre de driver inexistente, desactivar el cifrado, guardar contraseñas en el script, concatenar SQL, olvidar commit() y cargar millones de filas en memoria. También es común dejar conexiones abiertas o registrar secretos.
Conclusión
pyodbc permite conectar Python con SQL Server de forma directa. Instala el controlador correcto, guarda credenciales fuera del código, utiliza consultas parametrizadas y controla transacciones. Añade timeouts, logging seguro, permisos mínimos y pruebas en una base separada. Con estas prácticas puedes automatizar consultas e informes sin comprometer integridad ni seguridad.
