测试 Pyodbc 连接对象

Question

我有一个脚本，它在启动时启动与 Teradata 数据库的 Pyodbc 连接。 1 小时 20 分钟后，脚本重新使用连接对象对 Teradata 执行查询（使用 pandas）。查询非常简单，我知道它是正确的，因为我已经手动测试过它。当脚本执行它时，我收到以下错误消息：

oData = pandas.read_sql(oQuery, oConnexion)

Error message: (, DatabaseError("Execution failed on sql: SELECT DATA_QUALITY_INDICATOR\n ,DATA_QUALITY_INDICATOR_PARAMETER\n
,DATA_QUALITY_INDICATOR_PARAMETER_VALUE\n FROM AUDIT_VIEWS.AUDIT_DATA_QUALITY_INDICATOR_PARAMETER A\n JOIN AUDIT_VIEWS.AUDIT_DATA_QUALITY_INDICATOR B ON A.DATA_QUALITY_INDICATOR_KEY=B.DATA_QUALITY_INDICATOR_KEY\n
WHERE DATA_QUALITY_INDICATOR='DATA_COMPLETENESS_DNA_EXPORT_COUNTER_VS_CSV_DATA'\n AND DATA_QUALITY_INDICATOR_PARAMETER='oMeasures';\n('HY000', 'The driver did not supply an error!')\nunable to rollback",), )

虽然不是很明确，但我认为执行失败是因为连接对象可能已经超时（注意创建连接对象时没有设置超时）。你知道一个方法吗：

全程保持连接对象存活？
在执行查询之前测试连接对象是否仍然有效？

谢谢

Answer 1

我实施了以下解决方法，确认这是连接对象的超时问题。在重新使用我正在测试的连接对象之前，如果测试失败则重新打开连接。

 import pyodbc
 try:
 #test current connection object in global variable
      oConnexion = oGlobalVariable['oConnexion']
      oCursor = oConnexion.cursor()
      oCursor.execute("select '1'")
      oCursor.close()
 except:
      #if test fails re-open connection
      oConnexionString = fGetParameterFile(pFileName="global_parameter.ini", pSectionName="AUDIT", pParameterName="oConnexionString")
      oConnexion = pyodbc.connect(oConnexionString, autocommit=True)
      oConnexion.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
      oConnexion.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
      oConnexion.setdecoding(pyodbc.SQL_WMETADATA, encoding='utf-8')
      oConnexion.setencoding(encoding='utf-8')
      oGlobalVariable['oConnexion'] = oConnexion
      return oConnexion
 else:
      return oGlobalVariable['oConnexion']

测试 Pyodbc 连接对象

Test Pyodbc Connection Object

python

connection

timeout

pyodbc