buff3.0/buff-api/sync_goods/sync_db_errors.py

# -*- coding: utf-8 -*-
"""同步脚本共用的数据库连接类异常判断（用于日志后放行外层循环）。"""
import pymysql

# 连接失败、断连、服务端关闭、连接数过多等，适合记录日志后由定时循环重试
_DB_CONNECTION_ERR_CODES = frozenset(
    {
        2002,  # Can't connect to MySQL server
        2003,  # Can't connect to MySQL server (host)
        2006,  # MySQL server has gone away
        2013,  # Lost connection to MySQL server during query
        2014,  # Commands out of sync
        1053,  # Server shutdown in progress
        1040,  # Too many connections
    }
)


def is_database_connection_error(exc: BaseException) -> bool:
    """是否为连接/断连类问题（不中断 shell 循环，仅打日志）。"""
    if isinstance(exc, pymysql.err.InterfaceError):
        return True
    if isinstance(exc, pymysql.err.OperationalError):
        code = exc.args[0] if exc.args else None
        return code in _DB_CONNECTION_ERR_CODES
    return False