如何处理 Python 雪花连接器生成的异常/错误

Question

我正在尝试处理Python Snowflake连接器生成的异常/错误。

connection = snowflake.connector.connect(
user='user',
password=''password,
account='account',
role='ACCOUNTADMIN'
)

如果连接成功，那么它将返回一个雪花对象，否则我想处理所有可能的异常并打印自定义消息。

1. 雪花.连接器.连接方法将生成哪些可能的异常。

2. 如何处理所有这些异常。(链接到文档或示例将非常有帮助)

Answer 1

如果您查看snowflake.connector.errors文件，您可以看到雪花连接器可以返回的所有错误的列表。它没有说明snowflake.connector.connect可以返回的特定错误。以下是我怀疑可以返回的错误类型的列表：

class InterfaceError(Error)
class DatabaseError(Error)
class InternalError(DatabaseError)
class OperationalError(DatabaseError)
class InternalServerError(Error)
class ServiceUnavailableError(Error)
class GatewayTimeoutError(Error)
class ForbiddenError(Error)
class RequestTimeoutError(Error)
class BadRequest(Error)
class BadGatewayError(Error)
class MethodNotAllowed(Error)
class OtherHTTPRetryableError(Error)
class MissingDependencyError(Error)

与其尝试处理每个错误，不如只处理您知道该怎么处理的错误。例如，除非您知道如何修复它，否则您不会处理InternalServerError。相反，记录它并让它失败。

以下是如何捕获错误的用户名/密码的示例。然后，您可以要求用户重新输入其详细信息：

import os
import snowflake.connector
from snowflake.connector.errors import DatabaseError, ProgrammingError
snowflake_account = os.environ['SNOWFLAKE_ACCOUNT']
if __name__ == '__main__':
try:
con = snowflake.connector.connect(
user='bad username',       # <-------- Bad user
password='bad password',   # <-------- Bad pass
account=snowflake_account  # <-------- This is correct
)
except DatabaseError as db_ex:
if db_ex.errno == 250001:
print(f"Invalid username/password, please re-enter username and password...")
# code for user to re-enter username & pass
else:
raise
except Exception as ex:
# Log this
print(f"Some error you don't know how to handle {ex}")
raise
else:
try:
results = con.cursor().execute("select * from db.schema.table").fetchall()
print(results)
except ProgrammingError as db_ex:
print(f"Programming error: {db_ex}")
raise
finally:
con.close()

我还举了一个捕获ProgrammingError的示例，当您输入一些无效的 SQL 时可以引发该。

Answer 2

你可以在python中使用try.. catch方法

try:
self.mainsfconnection = sf.connect(
user=config.user,
password=config.password,
account=config.account,
warehouse=config.warehouse,
role=config.role,
database=config.database,
schema=config.schema)
# -- Snowflake : Default Session parameters
self.mainsfconnection.cursor().execute("USE ROLE {0}".format(config.role))
self.mainsfconnection.cursor().execute("USE WAREHOUSE {0}".format(config.warehouse))
self.mainsfconnection.cursor().execute("USE SCHEMA {0}.{1}".format(config.database, config.schema))
# -- Exception clause : snowflake
except sf.errors.ProgrammingError as e:
print('SQL Execu tion Error: {0}'.format(e.msg))
print('Snowflake Query Id: {0}'.format(e.sfqid))
print('Error Number: {0}'.format(e.errno))
print('SQL State: {0}'.format(e.sqlstate))

Answer 3

这是文档中一个非常好的例子，请注意，没有理由不将cur =和connection =移动到try命令下方(当然要正确缩进)：

https://docs.snowflake.com/en/user-guide/python-connector-example.html#handling-errors

# Catching the syntax error
cur = con.cursor()
try:
cur.execute("SELECT * FROM testtable")
except snowflake.connector.errors.ProgrammingError as e:
# default error message
print(e)
# customer error message
print('Error {0} ({1}): {2} ({3})'.format(e.errno, e.sqlstate, e.msg, e.sfqid))
finally:
cur.close()