用 MSSQL-Python 重试逻辑和连接弹性

暂时性故障是指通过mssql-python驱动连接SQL Server和Azure SQL时可能发生的临时错误。 这些错误通常会自行解决：

• 网络连接短暂中断。
• 服务器资源限制。
• Azure SQL 限制。
• 故障切换事件。

实现重试逻辑可以提升应用的可靠性，尤其是对于云托管数据库。

不要用重试来隐藏配置或编码错误。 数据库缺失、凭证错误或连接池耗尽需要修复，而不是再尝试一次。

通过池化和默认 msodbcsql18 提供者，驱动程序在验证重用连接时临时应用一个本地连接操作超时：即正登录超时，或当登录超时为零或未指定时，超时为15秒。 这并不是完整验证或签出的截止时间。 多个原生操作、清理和开启替代连接可能会增加耗时。 如果验证失败，驱动程序会尝试替换连接，而不是等待每一个剩余的空闲连接。 这种恢复不会重放你的语句，也不会重试连接池耗尽。 有关详细信息，请参阅连接池。

识别瞬态错误

mssql-python 不会在异常时将 SQL Server 引擎的错误编号作为属性暴露。 相反，驱动程序将 SQLSTATE 代码映射到一组固定的 PEP 249 例外子类（OperationalError、、 ProgrammingError、等）以及属性中的 driver_error 标准化英文文本。 将该组合作为瞬态分类的基础。

可靠的瞬态信号

以下 SQLSTATE 值会以 OperationalError 形式返回，表示某种值得重试的情况。 右侧列显示了司机设置的具体 driver_error 文字：

SQLSTATE	driver_error 文本	条件
HYT00	Timeout expired	语句级别的超时。
HYT01	Connection timeout expired	连接超时。
08001	Client unable to establish connection	无法打开连接。
08S01	Communication link failure	网络断线、服务器重置、TCP故障。
08007	Connection failure during transaction	交易中途连接中断。
40001	Serialization failure	死锁牺牲品。
40003	Statement completion unknown	交易状态不确定。

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

Azure SQL 限制（尽力而为）

Azure SQL 限流错误（40197、40501、40613、49918、49919、49920及相关代码）通常会以 SQLSTATE 42000 的形式出现，mssql-python 会将其映射为 ProgrammingError。 引擎错误编号不会作为属性显示，所以唯一的信号是属性中的 ddbc_error 服务器消息文本。

请检查 ddbc_error 中是否存在已知号码（如果有），以及标准的英文错误 40501 消息。 号码可能缺失，消息文本也不是稳定的约定。 局部化或格式不同的诊断可能不匹配：

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)

哪些内容不应重试

应快速失败而非重试的错误示例包括无效凭证（OperationalError 带驱动文本 Invalid authorization specification）、数据库不存在或权限不足、语法错误（ProgrammingError）、缺少对象以及连接池耗尽。 is_transient_error上述函数通过构造排除了所有这些。

基础重试装饰器

带有固定延迟的简单重试

一个装饰器，以固定延迟重试被包装的函数固定次数：

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()

指数退避

使用可选抖动以指数方式增加重试之间的延迟，以分散并发重试：

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()

连接重试类

弹性连接管理器

一个既能处理重试也处理自动重连的连接封装器：

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")

Azure SQL 特定处理

在 识别瞬态错误 中，使用 mssql-python 异常映射来分类以下目录中的错误。

以下错误在连接建立过程中或向服务器发送请求时可能为暂时性。 在短的、受限的退避时重试。 重复几次后依然存在的错误通常表明配置问题，比如服务器错误、权限缺失、加密设置不兼容或配额用尽，重试无法解决。

Error	失败类型	消息	Troubleshooting
64	登录时的传输	A connection was successfully established with the server, but then an error occurred during the login process. (provider: TCP Provider, error: 0 - The specified network name is no longer available.)	TCP 连接在握手过程中中断。 并非凭据错误。 如果问题仍然存在，请检查是否存在客户端网络不稳定的情况，或是否有会丢弃半建立连接的中间设备。
233	预登录传输（TLS）	The client was unable to establish a connection because of an error during connection initialization process before login.	当服务器因资源耗尽、连接限制或客户端不支持而无法接受连接时，通常会返回此错误。 并非凭据错误。 验证服务器运行状况，然后检查客户端登录超时、TLS 设置和客户端/服务器 TLS 版本兼容性。
4060	数据库可用性或访问	Cannot open database "%.*ls" requested by the login. The login failed.	登录已通过身份验证，但无法打开所请求的数据库。 暂时原因包括数据库处于转换过程中（故障转移、还原、扩缩容）或已自动暂停。 永久性原因（数据库不存在，登录缺少访问权限）不会通过重试来修复;检查数据库名称、登录映射和数据库状态。
4221	可读的次要过渡	Login to read-secondary failed due to long wait on 'HADR_DATABASE_WAIT_FOR_TRANSITION_TO_VERSIONING'.	该副本无法用于登录访问，因为在回收该副本时，当时正在进行中的事务缺少行版本。 回滚或提交主节点上的活动事务以解决该问题。 通过避免主数据库上的长时间写入事务来缓解此问题。
10053	本地传输中止	A transport-level error has occurred when sending the request to the server. (provider: TCP Provider, error: 0 - An established connection was aborted by the software in your host machine.)	本地端中止连接。 检查客户端网络运行状况以及任何本地防火墙或 VPN 客户端。
10054	重置远程传输	A transport-level error has occurred when sending the request to the server. (provider: TCP Provider, error: 0 - An existing connection was forcibly closed by the remote host.)	远程端发送 TCP 重置。 常见原因包括：对端进程崩溃、防火墙注入了重置包，或 Azure SQL 网关关闭了空闲连接。 对于空闲重置这类情况，请在客户端启用 TCP 保活，或缩短连接池的空闲超时时间。
10060	连接超时	A connection attempt failed because the connected party did not properly respond after a period of time.	服务器或中间网络设备在TCP连接超时前没有响应。 检查服务器健康状况、路由、防火墙规则，以及配置好的主机和端口是否可访问。
10928	数据库资源限制	Resource ID: %d. The %s limit for the database is %d and has been reached.	数据库超过Azure SQL资源治理限制。 资源 ID 1 表示工作线程限制；资源 ID 2 表示会话限制。 从消息中识别限制类型，然后降低并发、扩容数据库，或缩短长时间占用该资源的操作的执行时间。
10929	数据库节流	Resource ID: %d. The %s minimum guarantee is %d, maximum limit is %d, and the current usage for the database is %d. However, the server is currently too busy to support requests greater than %d for this database.	数据库已超过其最低保障值，底层服务器正在对其进行限流。 当邻居负载下降时，重试通常成功。 持续出现表示需要更高的服务层级或不太干扰的环境。
40020、40143、40166、40540	Azure SQL 故障转移子代码	在故障转移期间，于错误 40197 的 Error code %d 槽中报告。	40197 故障转移消息中嵌入了子代码，某些路径会将其作为顶级错误代码显示。 将它们视为 40197。
40197	Azure SQL 故障转移	The service has encountered an error processing your request. Please try again. Error code %d.	Azure SQL 中的一次软件升级、硬件故障或其他故障转移事件。 重新连接会将你路由到一个健康的副本。 嵌入的错误代码标识故障转移类型。 如果错误仍然存在，请捕获会话跟踪 ID 并联系支持人员。
40501	Azure SQL 限制	The service is currently busy. Retry the request after 10 seconds. Incident ID: %ls. Code: %d.	Azure SQL 引擎限流。 建议的最低退避时间为 10 秒。 持续限制表明工作负荷已超过数据库的资源分配；升级服务层级或减少并发。
40613	数据库不可用	Database '%.*ls' on server '%.*ls' is not currently available. Please retry the connection later. If the problem persists, contact customer support, and provide them with the session tracing ID of '%.*ls'.	数据库不可用，通常发生在故障转移过程中，或在缩放操作期间短暂出现。 按退避策略重试；如果问题持续几分钟以上，请记录会话跟踪 ID 并创建支持工单。
42108	SQL 池已暂停	Can not connect to the SQL pool since it is paused. Please resume the SQL pool and try again.	专用 SQL 池（Synapse）处于暂停状态。 只有在池恢复后，重试才会成功。 显式地恢复池，或将工作负载安排在池恢复后运行。
42109	SQL 池正在恢复	The SQL pool is warming up. Please try again.	专用 SQL 池正在恢复。 按退避策略重试，直到池上线；预热通常需要几分钟。
49918	服务资源短缺	Cannot process request. Not enough resources to process request. The service is currently busy. Please retry the request later.	服务器当前无法分配足够的资源来满足请求。 在退避时重试。 如果错误仍然存在，请纵向扩展数据库或弹性池。
49919	管理操作节流	Cannot process create or update request. Too many create or update operations in progress for subscription "%ld".	管理操作的订阅级并发限制。 减少并行创建/更新调用或将它们错开。
49920	订阅操作限制	Cannot process request. Too many operations in progress for subscription "%ld".	针对正在进行的操作的订阅级并发限制。 降低并行度，或等待当前正在进行的操作完成。

语句级错误不在此列表中，因为它们发生在连接建立之后，而且即使失败，会话仍可继续使用。 最常见的可重试语句错误是 1205（死锁受害者）和 1222（锁请求超时）。 重试整个事务，而不是单个失败语句。

错误消息文本来自Azure SQL暂时性连接错误。 这些错误可在 SQL Server、Azure SQL 数据库、Azure SQL 托管实例、Microsoft Fabric 中的 SQL 数据库以及 Azure Synapse Analytics 中的专用 SQL 池中重试。

处理 Azure 限制

Azure SQL 限流错误需要比标准瞬态错误更长的延迟和更多的重试。 从is_azure_throttling中重用：

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

处理故障转移

当 Azure SQL 或可用性组故障切换导致连接中断时，请重新连接并重试：

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"}
)

死锁处理

发生死锁时重试

死锁（错误1205）是暂时性的。 用一个短暂的随机延迟重试以打破僵局循环。 重试可以解决立即失败的问题，但反复出现的死锁说明设计存在问题，需要在服务器端调查。 有关分析和解决根本原因的指导，请参见 “死锁错误”。

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

结构化重试与配置

重试策略类

将重试配置封装在数据类中，以便在不同操作间重复使用：

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)

断路器设计模式

通过跟踪连续错误并在达到阈值时暂时阻断呼叫，防止连锁故障：

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

出现配置错误时不要重试

并非所有错误都是暂时的。 重试配置或编码错误浪费时间，可能掩盖真正的问题。 只重试那些可能自行解决的错误。 因为 mssql-python 不把引擎错误编号作为属性暴露，所以可以按异常子类加上 driver_error 文本分类。

切勿重试这些（ 改为修正代码或配置）：

条件	例外类型	driver_error 文本	修复
无效对象名（引擎208）	ProgrammingError	Base table or view not found	该表不存在。 修正查询或创建表。
无效的列名（引擎 207）	ProgrammingError	Column not found	那根柱子不存在。 检查架构。
语法错误（引擎102）	ProgrammingError	Syntax error or access violation	修复查询。
登录失败（发动机18456）	OperationalError	Invalid authorization specification	凭证错误。 修正连接字符串（连接字符串）。
数据库缺失或访问被拒（引擎 4060）	OperationalError	Server rejected the connection	数据库不存在，或者登录时无法访问。 固定目标或权限。
连接池耗尽	OperationalError	（变化）	增加连接池容量，及时释放连接，或降低并发度。
ConnectionStringParseError	独立	无	连接字符串关键字中的拼写错误。 修复该字符串。
不支持的功能	NotSupportedError	Optional feature not implemented	采用替代方法。

一定要重试这些（ 它们会自行解决）：

条件	例外类型	driver_error 文本
语句超时	OperationalError	Timeout expired
连接超时	OperationalError	Connection timeout expired
无法打开连接	OperationalError	Client unable to establish connection
网络掉落	OperationalError	Communication link failure
连接在事务中途断开	OperationalError	Connection failure during transaction
死锁牺牲品（引擎 1205）	OperationalError	Serialization failure
不确定交易状态	OperationalError	Statement completion unknown
Azure SQL throttling （40197， 40501， 40613， 49918–49920）	ProgrammingError	Syntax error or access violation （用于 is_azure_throttling 匹配 ddbc_error 文本）

相关内容

• mssql-python 的错误处理与 SQLSTATE 代码
• 使用 mssql-python 进行连接池管理
• mssql-python 故障排除
• 排查暂时性连接错误