Как проверить соединение в python?

Я делаю проект, считывая метку RFID, используя python на малине pi и используя считыватель RDM880.

Моя идея – потратить время и время, чтобы проверить работу персонала вовремя или нет.

Я пытаюсь добавить card_ID, time_in, time_out в локальную mysql и удаленную mysql (IP: 192.168.137.1) с помощью python.

Он имеет ту же таблицу в удаленном и локальном mysql.

Если mysql remote поврежден, я хочу добавить только локальный mysql.

Вот мой код:

import serial import time import RPi.GPIO as GPIO import MySQLdb from datetime import datetime from binascii import hexlify serial=serial.Serial("/dev/ttyAMA0", baudrate=9600, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout=0.1) db_local = MySQLdb.connect("localhost","root","root","luan_van") #connect local db = MySQLdb.connect("192.168.137.1", "root_a","","luan_van") #connect remote ID_rong = 128187 # reader respone if no card chuoi= "\xAA\x00\x03\x25\x26\x00\x00\xBB" def RFID(str): #function read RFID via uart serial.write(chuoi) data = serial.readline() tach_5 = data[5] tach_6 = data[6] hex_5 = hexlify(tach_5) hex_6= hexlify(tach_6) num_5 = int(hex_5,16) num_6 = int(hex_6,16) num_a = num_5 * 1000 + num_6 if(num_a != ID_rong): tach_7 = data[7] tach_8 = data[7] hex_7 = hexlify(tach_7) hex_8= hexlify(tach_8) num_7 = int(hex_7,16) num_8 = int(hex_8,16) num = num_8 + num_7 * 1000 + num_6 * 1000000 + num_5 * 1000000000 else: num = num_5 * 1000 + num_6 return num def add_database(): # add card_ID and time_in to remote mysql with db: cur = db.cursor() cur.execure("INSERT INTO tt_control(Card_ID,Time_in) VALUES ('%d',NOW()) " %num) return def add_database_local(): # add card_ID and time_in to remote mysql with db_local: cur = db_local.cursor() cur.execure("INSERT INTO tt_control(Card_ID,Time_in) VALUES ('%d',NOW()) " %num) return def have_ID(int): #check ID in table tt_control with db_local: cur = db_local.cursor(MySQLdb.cursors.DictCursor) cur.execute("SELECT * FROM tt_control WHERE Card_ID = '%d'" %num) rows = cur.fetchall() ID="" for row in rows: ID = row['Card_ID'] return ID def add_time_out(): #add time out to remote mysql with db: cur = db.cursor(MySQLdb.cursors.DictCursor) cur.execute("UPDATE tt_control SET Time_out = NOW() WHERE Card_ID = '%d'" %num) return def add_time_out_local(): #add time out to local mysql with db_local: cur = db_local.cursor(MySQLdb.cursors.DictCursor) cur.execute("UPDATE tt_control SET Time_out = NOW() WHERE Card_ID = '%d'" %num) return def add_OUT(): #increase Card_ID to distinguish second check with db: cur = db.cursor(MySQLdb.cursors.DictCursor) cur.execute("UPDATE tt_control SET Card_ID = Card_ID + 1 WHERE Card_ID = '%d'" %num) return def add_OUT_local(): #increase Card_ID to distinguish second check with db_local: cur = db_local.cursor(MySQLdb.cursors.DictCursor) cur.execute("UPDATE tt_control SET Card_ID = Card_ID + 1 WHERE Card_ID = '%d'" %num) return while 1: num = RFID(chuoi) time.sleep(1) Have_ID =have_ID(num) if(num != ID_rong): if(Have_ID ==""): add_database() #---> it will error if remote broken, how can i fix it? add_database_local() else: add_time_out() #---> it will error if remote broken, how can i fix it? I think connection alive can fix, but I don't know add_time_out_local() add_OUT() add_OUT_local() #---> it will error if remote broken, how can i fix it?

2 Solutions collect form web for “Как проверить соединение в python?”

У вас есть пара вариантов:

(не так хорошо) Пинг сервера регулярно, чтобы поддерживать связь.

(лучше всего) Обрабатывать исключение MySQLdb при вызове cur.execute путем повторного установления соединения и повторного вызова вызова. Вот отличный и краткий ответ о том, как это сделать . Из этой статьи вы сами справляетесь с этим исключением:

 def __execute_sql(self,sql,cursor): try: cursor.execute(sql) return 1 except MySQLdb.OperationalError, e: if e[0] == 2006: self.logger.do_logging('info','DB', "%s : Restarting db" %(e)) self.start_database() return 0

(наконец) Установите новое соединение с базой данных непосредственно перед тем, как вы действительно вызовите записи базы данных. В этом случае переместите определения db и db_local в функцию, которую вы вызываете непосредственно перед cursor . Если вы делаете тысячи запросов, это не самое лучшее. Однако, если это всего лишь несколько запросов к базе данных, это, вероятно, хорошо.

Я использую следующий метод:

 def checkConn(self): sq = "SELECT NOW()" try: self.cur.execute( sq ) except pymysql.Error as e: if e.errno == 2006: return self.connect() else: print ( "No connection with database." ) return False