Python中如何检查pyodbc数据库连接是否正常？

使用 pyodbc 长时间连接一个数据库，但试验了下隔段时间查询就会报 01000 的错，怎么在查询前检查连接是否正常呢？


Python中如何检查pyodbc数据库连接是否正常？
1 回复 

import pyodbc

def check_db_connection(connection_string):
    """
    检查pyodbc数据库连接是否正常
    
    参数:
        connection_string: 数据库连接字符串
        
    返回:
        tuple: (是否成功, 错误信息/成功消息)
    """
    connection = None
    try:
        # 尝试建立连接
        connection = pyodbc.connect(connection_string)
        
        # 执行简单的测试查询
        cursor = connection.cursor()
        cursor.execute("SELECT 1")
        result = cursor.fetchone()
        
        if result and result[0] == 1:
            return True, "数据库连接正常"
        else:
            return False, "测试查询失败"
            
    except pyodbc.Error as e:
        # 捕获pyodbc特定的错误
        return False, f"数据库连接失败: {str(e)}"
        
    except Exception as e:
        # 捕获其他可能的错误
        return False, f"发生未知错误: {str(e)}"
        
    finally:
        # 确保连接被关闭
        if connection:
            connection.close()

# 使用示例
if __name__ == "__main__":
    # 示例连接字符串（根据你的数据库修改）
    conn_str = (
        "DRIVER={ODBC Driver 17 for SQL Server};"
        "SERVER=your_server;"
        "DATABASE=your_database;"
        "UID=your_username;"
        "PWD=your_password"
    )
    
    # 或者使用信任连接
    # conn_str = (
    #     "DRIVER={ODBC Driver 17 for SQL Server};"
    #     "SERVER=your_server;"
    #     "DATABASE=your_database;"
    #     "Trusted_Connection=yes;"
    # )
    
    success, message = check_db_connection(conn_str)
    print(f"成功: {success}, 消息: {message}")

这个函数做了几件事：

  1. 尝试用给定的连接字符串建立连接
  2. 执行一个简单的SELECT 1查询来验证连接确实可用
  3. 妥善处理各种异常情况
  4. 确保连接被正确关闭

关键点：

  • 使用try-except捕获pyodbc.Error和其他异常
  • 执行测试查询确保连接不只是建立，还能正常工作
  • 在finally块中关闭连接，避免资源泄漏

实际使用时，把连接字符串换成你自己的配置就行。不同的数据库（SQL Server、MySQL、PostgreSQL等）连接字符串格式会有所不同。

简单说就是：连一下，查一下，有问题就报错。

回到顶部