使用pandas读SQL时如何捕获pyodbc的DataError异常？

当使用SQLAlchemy引擎配合pandas.read_sql时，直接捕获pyodbc.DataError会失败，核心原因是SQLAlchemy会将数据库驱动抛出的异常包装成自身的异常类，实际抛出的是sqlalchemy.exc.DataError，而非原始的pyodbc异常。

解决方案一：精准捕获pyodbc引发的DataError #

如果需要明确识别是pyodbc的DataError导致的问题，可以先捕获SQLAlchemy的异常，再检查其内部的原始异常：

import pandas as pd 
import pyodbc
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.exc import DataError as SADataError  # 导入SQLAlchemy的DataError

driver = "SQL Server"
server = "my_server" 
database = "my_db" 

connection_string = (f"Driver={driver};"
                     f"Server={server};"
                     f"Database={database};"
                     f"Trusted_Connection=yes;"
                     f"TrustServerCertificate=yes")

connection_url = URL.create("mssql+pyodbc", 
                            query={"odbc_connect": connection_string})

engine = create_engine(connection_url)

query = (f"SELECT top 10 * "
         f"FROM abc.table_name")

try: 
    df = pd.read_sql(sql=query, con=engine)
except SADataError as sa_err:
    # 验证原始异常是否为pyodbc.DataError
    if isinstance(sa_err.orig, pyodbc.DataError):
        print(f"捕获到pyodbc数据转换错误: {sa_err.orig}")
        # 此处可添加自定义处理逻辑，比如记录日志、跳过错误行等
    else:
        print(f"其他SQLAlchemy数据错误: {sa_err}")

解决方案二：直接捕获SQLAlchemy的DataError #

如果不需要区分具体驱动的异常，仅需处理所有数据转换类错误，可以直接捕获SQLAlchemy的DataError：

from sqlalchemy.exc import DataError

try: 
    df = pd.read_sql(sql=query, con=engine)
except DataError as err:
    print(f"数据转换错误: {err}")
    # 添加自定义处理逻辑

额外优化建议 #

你的错误本质是数据库表中某列定义为int类型，但实际存储了'00:35'这类nvarchar值，导致驱动转换失败。如果业务允许，也可以在SQL查询阶段提前处理异常值：

• 使用TRY_CAST（SQL Server 2012+支持）：

SELECT top 10 
    TRY_CAST(问题列名 AS INT) AS 问题列名,
    -- 其他列...
FROM abc.table_name

• 或用CASE判断：

SELECT top 10 
    CASE WHEN ISNUMERIC(问题列名) = 1 THEN CAST(问题列名 AS INT) ELSE NULL END AS 问题列名,
    -- 其他列...
FROM abc.table_name

这种方式能从源头避免驱动层面的转换错误。

内容的提问来源于stack exchange，提问作者marianoju