Identifying sqlalchemy.exc.OperationalError

user_4866865 0
Viewed 3690

I'm trying to catch mysql/sqlalchemy OperationalErrors and replace handle access denied (1045) differently from connection refused (2003)

sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1045, "Access denied for user … (Background on this error at: http://sqlalche.me/e/e3q8)
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on 'localhost' ([Errno 111] Connection refused)") (Background on this error at: http://sqlalche.me/e/e3q8)

I just can't seem to find any documentation on how to tell these apart programmatically. I dived into the sources and thought I could check the value of err.orig.original_exception.errno but that was not the case.

Edit: err.orig doesn't seem to be defined for access denied which might be a bug.

try:
  engine.scalar(select([1]))
except sqlalchemy.exc.OperationalError as err:
  if err_______:
    print("Access Denied")
  elifif err_______:
    print("Connection Refused")
  else:
    raise

This issue really bugs me and even the bounty is running out with no news. I'm starting to believe it must be a bug in sqlalchemy but the sqlalchemy documentation is not very descriptive in that regard and I'm new to sqlalchemy and python in general so it's really hard for me to tell. I couldn't find support on irc either, where do I go from here?

python-db-api sqlalchemy python
2 Answers

After some more research, I found the mysql error code to be in err.orig.args[0]. So the Answer is:

try:
  engine.scalar(select([1]))
except sqlalchemy.exc.OperationalError as err:
  if err.orig.args[0]==1045:
    print("Access Denied")
  elif err.orig.args[0]==2003:
    print("Connection Refused")
  else:
    raise
user_48668650

Try err.args[0]

try:
  engine.scalar(select([1]))
except sqlalchemy.exc.OperationalError as err:
  if err.args[0] == 1045:
    print("Access Denied")
  elif err.args[0] == 2003:
    print("Connection Refused")
  else:
    raise

This should be what you're looking for. Refer to the documentation for more reading

Edit

As OperationalError wraps DBAPIError, that has a code argument. Most likely just replace args[0] with code. Like so:

try:
  engine.scalar(select([1]))
except sqlalchemy.exc.OperationalError as err:
  if err.code == 1045:
    print("Access Denied")
  elif err.code == 2003:
    print("Connection Refused")
  else:
    raise
user_2250200
Related