Lógica de reintentos y resiliencia de conexión con mssql-python

Los fallos transitorios son errores temporales que pueden ocurrir al conectarse a SQL Server y Azure SQL a través del controlador mssql-python. Estos errores a menudo se resuelven por sí solos:

• La conectividad de red presenta interrupciones momentáneas.
• Limitaciones de recursos del servidor.
• Limitación de Azure SQL.
• Eventos de conmutación por error.

Implementar lógica de reintentos mejora la fiabilidad de las aplicaciones, especialmente para bases de datos alojadas en la nube.

No uses reintentos para ocultar errores de configuración o de programación. Una base de datos ausente, credenciales defectuosas o un conjunto de conexiones agotado necesitan una solución, no otro intento.

Con el pooling y el proveedor por defecto msodbcsql18 , el controlador aplica temporalmente un tiempo de espera nativo para la operación de conexión durante la validación de una conexión reutilizada: el tiempo de espera positivo de inicio de sesión, o 15 segundos cuando el tiempo de espera de inicio de sesión es cero o no se especifica. Esto no es una fecha límite total de validación ni de cierre. Múltiples operaciones nativas, limpieza y apertura de una conexión de reemplazo pueden añadir tiempo. Si la validación falla, el controlador intenta un reemplazo en lugar de esperar en cada conexión inactiva restante. Esta recuperación no repite tus instrucciones ni reintenta el agotamiento del grupo de reintentos. Para más información, consulte Agrupación de conexiones.

Identificar errores transitorios

MSSQL-Python no expone el número de error del motor SQL Server como un atributo en las excepciones. En su lugar, el controlador asigna códigos SQLSTATE a un conjunto fijo de subclases de excepción PEP 249 (OperationalError, ProgrammingError, y así sucesivamente) y a texto estandarizado en inglés en el driver_error atributo. Usa esa combinación como base para la clasificación de transitorios.

Señales transitorias fiables

Los siguientes valores de SQLSTATE aparecen como OperationalError e indican una condición que merece la pena volver a intentar. La columna de la derecha muestra el texto exacto driver_error establecido por el conductor:

import mssql_python


TRANSIENT_DRIVER_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
    "Serialization failure",
    "Statement completion unknown",
})


def is_transient_error(error: BaseException) -> bool:
    """Return True if the exception represents a retryable transient failure.

    Classification is based on the driver's PEP 249 exception subclass and
    on the standardized `driver_error` text that mssql-python sets from
    the SQLSTATE returned by the server.
    """
    if isinstance(error, mssql_python.OperationalError):
        return getattr(error, "driver_error", "") in TRANSIENT_DRIVER_ERRORS
    return False

Limitación de Azure SQL (máximo esfuerzo)

Los errores de limitación de Azure SQL (40197, 40501, 40613, 49918, 49919, 49920 y códigos relacionados) suelen aparecer con SQLSTATE 42000, que mssql-python asigna a ProgrammingError. El número de error del motor no aparece como un atributo, así que la única señal es el texto del mensaje del servidor en el ddbc_error atributo.

Comprueba ddbc_error si hay números conocidos y el mensaje estándar de error 40501 en inglés. El número puede estar ausente y el texto del mensaje no es un contrato estable. Los diagnósticos localizados o con formato diferente pueden no coincidir:

import re

# Azure SQL throttling and reconfiguration error numbers.
AZURE_THROTTLING_ERRORS = frozenset({
    40197, 40501, 40540, 40613, 40680, 49918, 49919, 49920, 10928, 10929,
})

_ERROR_NUMBER_RE = re.compile(r"\b(?:Error|Msg)\s+(\d+)\b")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


def is_azure_throttling(error: BaseException) -> bool:
    """Best-effort detection of Azure SQL throttling in ProgrammingError text."""
    if not isinstance(error, mssql_python.ProgrammingError):
        return False
    ddbc_text = getattr(error, "ddbc_error", "") or ""
    return (
        _AZURE_40501_MESSAGE in ddbc_text
        or any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))
    )


def is_retryable(error: BaseException) -> bool:
    return is_transient_error(error) or is_azure_throttling(error)

Qué no volver a intentar

Ejemplos de errores que deberían fallar rápido en lugar de reintentarlo incluyen credenciales inválidas (OperationalError con texto Invalid authorization specificationdel driver), una base de datos faltante o permisos insuficientes, errores de sintaxis (ProgrammingError), objetos ausentes y agotamiento del pool de conexión. La función is_transient_error anterior excluye a todos ellos por definición.

Decorador básico de reensayos

Reintento sencillo con retardo fijo

Un decorador que reintenta la función envuelta un número fijo de veces con un retraso constante:

import time
import functools
import mssql_python

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """Decorator to retry database operations on transient failures."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except mssql_python.Error as e:
                    last_exception = e
                    if not is_transient_error(e) or attempt == max_retries:
                        raise
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

# Usage
@retry_on_failure(max_retries=3, delay=2.0)
def get_user(cursor, user_id: int):
    cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = %(id)s", {"id": user_id})
    return cursor.fetchone()

Retroceso exponencial

Aumenta exponencialmente el retardo entre intentos con jitter opcional para repartir los intentos simultáneos:

import time
import random

def retry_with_backoff(max_retries: int = 5, 
                       base_delay: float = 1.0,
                       max_delay: float = 30.0,
                       jitter: bool = True):
    """Retry with exponential backoff and optional jitter."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except mssql_python.Error as e:
                    last_exception = e
                    if not is_transient_error(e) or attempt == max_retries:
                        raise
                    
                    # Calculate delay with exponential backoff
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    if jitter:
                        delay = delay * (0.5 + random.random())
                    
                    print(f"Attempt {attempt + 1} failed. Retrying in {delay:.2f}s...")
                    time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

@retry_with_backoff(max_retries=5, base_delay=1.0, max_delay=30.0)
def execute_query(cursor, query: str, params: dict):
    cursor.execute(query, params)
    return cursor.fetchall()

Clase de reintento de conexión

Gestor de conexiones resiliente

Un envoltorio de conexión que gestiona tanto reintentos como reconexiones automáticas:

import mssql_python
import time
import logging

# This example uses is_transient_error from the "Identify transient errors"
# section earlier in this article. Include that helper in your module.

# Configure logging so the retry and reconnect messages are visible
logging.basicConfig(level=logging.INFO)

class ResilientConnection:
    """Connection wrapper with automatic retry and reconnection."""
    
    def __init__(self, connection_string: str, max_retries: int = 5,
                 base_delay: float = 1.0, max_delay: float = 60.0):
        self.connection_string = connection_string
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self._conn = None
        self._logger = logging.getLogger(__name__)
    
    def _connect(self) -> mssql_python.Connection:
        """Establish connection with retry logic."""
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                self._logger.debug(f"Connection attempt {attempt + 1}")
                return mssql_python.connect(self.connection_string)
            except mssql_python.Error as e:
                last_exception = e
                if not is_transient_error(e) or attempt == self.max_retries:
                    self._logger.error(f"Connection failed: {e}")
                    raise
                
                delay = min(self.base_delay * (2 ** attempt), self.max_delay)
                self._logger.warning(f"Connection attempt {attempt + 1} failed. "
                                   f"Retrying in {delay:.1f}s...")
                time.sleep(delay)
        
        raise last_exception
    
    @property
    def connection(self) -> mssql_python.Connection:
        """Get or create connection."""
        if self._conn is None:
            self._conn = self._connect()
        return self._conn
    
    def execute(self, query: str, params: dict = None):
        """Execute query with automatic retry and reconnection."""
        return self._execute_with_retry(
            lambda c: self._do_execute(c, query, params)
        )
    
    def _do_execute(self, cursor, query: str, params: dict):
        cursor.execute(query, params or {})
        return cursor.fetchall()
    
    def _execute_with_retry(self, operation):
        """Execute an operation with retry logic."""
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                cursor = self.connection.cursor()
                return operation(cursor)
            except mssql_python.Error as e:
                last_exception = e
                
                if not is_transient_error(e):
                    raise
                
                if attempt == self.max_retries:
                    raise
                
                # Try to reconnect
                self._logger.warning(f"Operation failed. Reconnecting...")
                self._close()
                
                delay = min(self.base_delay * (2 ** attempt), self.max_delay)
                time.sleep(delay)
        
        raise last_exception
    
    def _close(self):
        """Close connection."""
        if self._conn:
            try:
                self._conn.close()
            except:
                pass
            self._conn = None
    
    def close(self):
        """Public close method."""
        self._close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False

# Usage
with ResilientConnection(connection_string) as db:
    users = db.execute("SELECT * FROM Person.Person WHERE EmailPromotion = %(promo)s", 
                       {"promo": 1})
    print(f"Retrieved {len(users)} rows")

Manejo específico de Azure SQL

Utiliza el mapeo de excepciones mssql-python en Identificar errores transitorios para clasificar errores del siguiente catálogo.

Los siguientes errores pueden ser transitorios cuando ocurren durante el establecimiento de la conexión o al enviar una solicitud al servidor. Vuelva a intentarlo en un retroceso corto y limitado. Los errores que persisten después de unos pocos intentos suelen indicar un problema de configuración, como un servidor equivocado, permisos faltos, ajustes de cifrado incompatibles o cuota agotada, que intentar de nuevo no solucionará.

Los errores a nivel de instrucción no están en esta lista porque se producen después de que se haya establecido la conexión y el fallo no impide seguir usando la sesión. Los errores de instrucciones que se pueden reintentar más comunes son 1205 (víctima de interbloqueo) y 1222 (tiempo de espera de la solicitud de bloqueo). Reintente la transacción completa en lugar de la sola instrucción que falla.

El texto del mensaje de error procede de Azure SQL errores de conexión transitorios. Estos errores se pueden volver a intentar en SQL Server, Azure SQL Database, Azure SQL Managed Instance, la base de datos SQL de Microsoft Fabric y los pools SQL dedicados de Azure Synapse Analytics.

Administrar la limitación de Azure

Los errores por limitación de Azure SQL requieren esperas más prolongadas y más reintentos que los errores transitorios estándar. Reutilizar is_azure_throttling de Identificar errores transitorios:

def execute_with_throttle_handling(cursor, query: str, params: dict,
                                   max_retries: int = 10,
                                   base_delay: float = 10.0):
    """Execute with extended retry for Azure SQL throttling."""
    for attempt in range(max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            if is_azure_throttling(e):
                if attempt < max_retries:
                    # Longer delays for throttling
                    delay = base_delay * (2 ** min(attempt, 4))  # Cap at 16 times the base delay
                    print(f"Throttled. Waiting {delay}s before retry...")
                    time.sleep(delay)
                    continue
            raise

Control de la conmutación por error

Vuelva a conectarse y a intentarlo cuando Azure SQL o la conmutación por error del grupo de disponibilidad interrumpan una conexión:

def execute_with_failover_retry(connect, query: str, params: dict,
                                max_retries: int = 3,
                                recovery_delay: float = 10.0):
    """Reconnect and retry during Azure SQL failover scenarios."""
    failover_numbers = frozenset({40613, 40197, 40540})
    last_exception = None

    for attempt in range(max_retries + 1):
        conn = None
        try:
            conn = connect()
            cursor = conn.cursor()
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            last_exception = e

            # Failover surfaces either as a transient OperationalError or as
            # a ProgrammingError whose ddbc_error text contains the engine
            # error number. Treat both as recoverable.
            ddbc_text = getattr(e, "ddbc_error", "") or ""
            is_failover = is_transient_error(e) or any(
                int(m) in failover_numbers for m in _ERROR_NUMBER_RE.findall(ddbc_text)
            )

            if is_failover and attempt < max_retries:
                print(f"Failover detected. Reconnecting in {recovery_delay}s...")
                if conn is not None:
                    try:
                        conn.close()
                    except mssql_python.Error:
                        pass
                time.sleep(recovery_delay)
                continue
            raise

    raise last_exception


# Usage
connection_string = (
    "Server=tcp:<server>.database.windows.net,1433;"
    "Database=AdventureWorks2025;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;TrustServerCertificate=no"
)

rows = execute_with_failover_retry(
    lambda: mssql_python.connect(connection_string),
    "SELECT TOP 10 ProductID, Name FROM Production.Product WHERE Color = %(color)s",
    {"color": "Silver"}
)

Administración de interbloqueos

Reintentar en caso de interbloqueo

Los bloqueos (error 1205) son transitorios. Inténtalo de nuevo con un breve retraso aleatorio para romper el ciclo de bloqueo. Reintentar soluciona el fallo inmediato, pero los bloqueos recurrentes indican un problema de diseño que deberías investigar en el servidor. Para orientación sobre cómo analizar y resolver la causa raíz, véase Errores de bloqueo.

def execute_with_deadlock_retry(cursor, query: str, params: dict,
                                max_retries: int = 3):
    """Automatically retry deadlocked transactions.

    Deadlocks (SQL Server error 1205) surface as OperationalError with
    driver_error == "Serialization failure" (SQLSTATE 40001).
    """
    for attempt in range(max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.OperationalError as e:
            if getattr(e, "driver_error", "") == "Serialization failure":
                if attempt < max_retries:
                    delay = random.uniform(0.1, 0.5) * (attempt + 1)
                    print(f"Deadlock detected. Retry {attempt + 1} in {delay:.2f}s")
                    time.sleep(delay)
                    continue
            raise

# Usage in transaction
conn.autocommit = False
try:
    cursor = conn.cursor()
    rows = execute_with_deadlock_retry(
        cursor,
        "SELECT TOP 5 Name, ListPrice FROM Production.Product WHERE ListPrice > %(price)s",
        {"price": 100}
    )
    conn.commit()
except Exception:
    conn.rollback()
    raise

Reintento estructurado con configuración

Clase de directiva de reintentos

Encapsular la configuración de reintentos en una clase de datos para su reutilización entre diferentes operaciones:

from dataclasses import dataclass, field
from typing import FrozenSet
import time
import random

# This example uses TRANSIENT_DRIVER_ERRORS from the "Identify transient errors"
# section earlier in this article. Include that allowlist in your module.

@dataclass
class RetryPolicy:
    """Configuration for retry behavior."""
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0
    exponential_base: float = 2.0
    jitter: bool = True
    transient_driver_errors: FrozenSet[str] = field(default_factory=lambda: TRANSIENT_DRIVER_ERRORS)

    def get_delay(self, attempt: int) -> float:
        """Calculate delay for given attempt number."""
        delay = min(
            self.base_delay * (self.exponential_base ** attempt),
            self.max_delay,
        )
        if self.jitter:
            delay *= (0.5 + random.random())
        return delay

    def should_retry(self, error: BaseException, attempt: int) -> bool:
        """Determine if operation should be retried."""
        if attempt >= self.max_retries:
            return False
        if isinstance(error, mssql_python.OperationalError):
            return getattr(error, "driver_error", "") in self.transient_driver_errors
        return False

def execute_with_policy(cursor, query: str, params: dict,
                        policy: RetryPolicy = None):
    """Execute query with configurable retry policy."""
    policy = policy or RetryPolicy()
    last_exception = None

    for attempt in range(policy.max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            last_exception = e
            if not policy.should_retry(e, attempt):
                raise

            delay = policy.get_delay(attempt)
            time.sleep(delay)

    raise last_exception

# Usage with custom policy
aggressive_retry = RetryPolicy(max_retries=10, base_delay=0.5, max_delay=60.0)
conservative_retry = RetryPolicy(max_retries=2, base_delay=5.0, max_delay=10.0)

results = execute_with_policy(cursor, query, params, aggressive_retry)

Patrón de disyuntor

Prevenir fallos en cascada rastreando errores consecutivos y bloqueando temporalmente las llamadas cuando se alcanza un umbral:

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject all calls
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    """Circuit breaker to prevent cascading failures."""
    
    def __init__(self, failure_threshold: int = 5,
                 recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self._lock = Lock()
    
    def can_execute(self) -> bool:
        """Check if circuit allows execution."""
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True
            
            if self.state == CircuitState.OPEN:
                # Check if recovery timeout has passed
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    return True
                return False
            
            # HALF_OPEN: allow one test request
            return True
    
    def record_success(self):
        """Record successful operation."""
        with self._lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED
    
    def record_failure(self):
        """Record failed operation."""
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

# Usage
circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)

def execute_with_circuit_breaker(cursor, query: str, params: dict):
    if not circuit.can_execute():
        raise Exception("Circuit breaker is open")
    
    try:
        cursor.execute(query, params)
        result = cursor.fetchall()
        circuit.record_success()
        return result
    except mssql_python.Error as e:
        if is_transient_error(e):
            circuit.record_failure()
        raise

No vuelvas a intentar los errores de configuración

No todos los errores son transitorios. Volver a intentar un error de configuración o código hace perder tiempo y puede enmascarar el problema real. Solo errores de reintento que puedan resolverse por sí solos. Como mssql-python no expone el número de error del motor como un atributo, clasifica por subclase de excepción más el driver_error texto.

Nunca vuelvas a intentar estos (arregla el código o la configuración en su lugar):

Reintenta siempre estos (se resuelven por sí solos):

Contenido relacionado

• Gestión de errores y códigos SQLSTATE para mssql-python
• Agrupación de conexiones con mssql-python
• Solucionar problemas en mssql-python
• Solución de problemas de errores de conexión transitorios