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?
Since SQLAlchemy relies on the DBAPI specification to interact with databases, the most common database management systems available are supported. PostgreSQL, MySQL, Oracle, Microsoft SQL Server, and SQLite are all examples of engines that we can use alongside with SQLAlchemy.
The SQLAlchemy engine creates a common interface to the database to execute SQL statements. It does this by wrapping a pool of database connections and a dialect in such a way that they can work together to provide uniform access to the backend database.
SQLAlchemy is a SQL tool built with Python that provides developers with an abundance of powerful features for designing and managing high-performance databases. We'll briefly explore how to use SQLAlchemy and then dive deeper into how to execute raw SQL statements from within the comfort of the Python domain language.
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
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With