I am getting the error InterfaceError (0, '')
. Is there way in Pymysql
library I can check whether connection or cursor is closed. For cursor I am already using context manager like that:
with db_connection.cursor() as cursor:
....
PyMySQL fetchAll The fetchall function gets all records. It returns a result set. Technically, it is a tuple of tuples. Each of the inner tuples represent a row in the table.
class pymysql.cursors. DictCursor (connection) A cursor which returns results as a dictionary.
It is not thread-safe, so call it before threads are created, or protect the call with a mutex. Arrange for mysql_thread_init() to be called early in the thread handler before calling any MySQL function. (If you call mysql_init() , it calls mysql_thread_init() for you.)
You can use Connection.open attribute.
The Connection.open field will be 1 if the connection is open and 0 otherwise. So you can say
if conn.open:
# do something
The conn.open attribute will tell you whether the connection has been explicitly closed or whether a remote close has been detected. However, it's always possible that you will try to issue a query and suddenly the connection is found to have given out - there is no way to detect this ahead of time (indeed, it might happen during the process of issuing the query), so the only truly safe thing is to wrap your calls in a try/except block
Use conn.connection in if statement.
import pymysql
def conn():
mydb=pymysql.Connect('localhost','root','password','demo_db',autocommit=True)
return mydb.cursor()
def db_exe(query,c):
try:
if c.connection:
print("connection exists")
c.execute(query)
return c.fetchall()
else:
print("trying to reconnect")
c=conn()
except Exception as e:
return str(e)
dbc=conn()
print(db_exe("select * from users",dbc))
This is how I did it, because I want to still run the query even if the connection goes down:
def reconnect():
mydb=pymysql.Connect(host='localhost',user='root',password='password',database='demo_db',ssl={"fake_flag_to_enable_tls":True},autocommit=True)
return mydb.cursor()
try:
if (c.connection.open != True):
c=reconnect() # reconnect
if c.connection.open:
c.execute(query)
return c.fetchall()
except Exception as e:
return str(e)
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