Is it possible to create a DB connection in all Python files Telegram Bot?

Hello!

Python not so long ago studied, there is a question...

Is it possible to create a DB connection (MySQL) in all Python files Telegram Bot? (In this case, three files)

Currently struggling with this:
pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')

Struggling like so:
# if the connection was lost, then it reconnects
db.ping(reconnect=True)
db.query('SET GLOBAL connect_timeout=86000')

if db.open is False:
 max_try = 15
 tryt = 0
 while db.open is False:
 if tryt < max_try:
 db.ping() # autoreconnect is true by default
 tryt +=1


db = pymysql.connect(host='localhost', ... the connection.
And so in three files.

Thanks for any help, hint!
April 19th 20 at 12:07

Related questions

More answers about "Is it possible to create a DB connection in all Python files Telegram Bot?"
1 answer
April 19th 20 at 12:09
Answering the main question, Yes, create connection in all files. But I would recommend the following approach:
from pymysql import MySQLError, connect
from pymysql.cursors import DictCursor


# Storing object with the connection in one file.
connection = connect(host='localhost', user='user', password='passwd', db='db',
 charset='utf8mb4', cursorclass=DictCursor)


def check_and_select_something():
 """The function of the first file with error checking and execution logic."""
try:
 with connection.cursor() as cursor:
 cursor.execute('SELECT *')
connection.commit()
 except MySQLError mysql_error as:
 print(f something went wrong. Error: {mysql_error}')
finally:
connection.close()


def select_something():
"""
 Funkcja the first file.

 Tests the connection, rebuilds if necessary, and causes
 the essential function that encapsulates the error checking
 at run time.
"""
connection.ping()
check_and_select_something()


def delete_something():
 """Funkcija second file (similar to first)."""


def insert_something():
 """Funkcja third of the file (similar to first)."""
Something went wrong. Error: (1096, 'No tables used')

check_and_select_something() here to stream all your samples from the base? - Jose_Cole73 commented on April 19th 20 at 12:12
@Jose_Cole73, All of these functions as an example; I didn't adjust Your code (especially that it is almost not there).

I guess I confused You.... Starting from the initial question, You can create a connection to the database multiple times in Your file; but I have suggested to keep the variable with the connection somewhere in one file and whenever it is used to perform the connection.ping() at the time of the execution of the required queries (that is, instead of the cursor.execute('SELECT *') in the example above You describe your query: cursor.execute("Your SQL query")) to handle exceptions. - Hellen_Leuschke6 commented on April 19th 20 at 12:15
@Hellen_Leuschke6, wrote the wrapper.
Can see how it will save!? Please! :)

import pymysql

class DBClient:

 def __init__(self, db_cfg):
 self.host = db_cfg['host']
 self.port = db_cfg['port']
 self.username = db_cfg['username']
 self.password = db_cfg['password']
 self.db_name = db_cfg['db_name']
 self.conn = pymysql.connect(host=self.host
port=self.port
user=self.username
password=self.password
db=self.db_name,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor)
 self.cur = self.conn.cursor()

 def execute(self, query, val = 0):
self.conn.ping()
try:
 with self.conn.cursor() as cursor:
 self.cur.execute(query, val)
self.conn.commit()
 except MySQLError mysql_error as:
 print(f something went wrong. Error: {mysql_error}')
finally:
self.conn.close()
#self.cur.execute(query)
#self.conn.commit()
 return self.cur.fetchall()

 def executeInsert(self, query, val = 0):
self.conn.ping()
try:
 with self.conn.cursor() as cursor:
 self.cur.execute(query, val)
self.conn.commit()
 except MySQLError mysql_error as:
 print(f something went wrong. Error: {mysql_error}')
finally:
self.conn.close()
#self.cur.execute(query)
#self.conn.commit()
 return self.cur.lastrowid

 def executeRowCount(self, query, val = "):
self.conn.ping()
try:
 with self.conn.cursor() as cursor:
 self.cur.execute(query, val)
self.conn.commit()
 except MySQLError mysql_error as:
 print(f something went wrong. Error: {mysql_error}')
finally:
self.conn.close()
#self.cur.execute(query)
#self.conn.commit()
 return self.cur.rowcount

 def executeAllUser(self, query):
self.conn.ping()
try:
 with self.conn.cursor() as cursor:
self.cur.execute(query)
self.conn.commit()
 except MySQLError mysql_error as:
 print(f something went wrong. Error: {mysql_error}')
finally:
self.conn.close()
#self.cur.execute(query)
#self.conn.commit()
 return self.cur.rowcount

 def close(self):
self.cur.close()
 self.conn.close()


Plug-in

cfg = dict()
cfg['host'] = 'localhost'
cfg['username'] = 'admin'
cfg['db_name'] = 'book'
cfg['password'] = '11111'
cfg['port'] = 3306

DBC = DBClient(cfg)


Use

sql = "SELECT * FROM db_users WHERE chat_id=%s"
val = (chat_id)
row = DBC.execute(sql, val)
- Jose_Cole73 commented on April 19th 20 at 12:18

Find more questions by tags TelegramPython