У меня есть класс, в котором я определяю соединение MySQL.Проблема в том, что я не могу использовать экземпляр класса, потому что соединение MySQL, кажется, уже закрыто.
Вот мой класс:
import mysql.connector
import sshtunnel
class DB:
def __init__(self):
self.pem = '...'
self.config = {
...
}
sshtunnel.SSH_TIMEOUT = 10.0
sshtunnel.TUNNEL_TIMEOUT = 10.0
with sshtunnel.SSHTunnelForwarder(
ssh_address = ('xx', yy),
ssh_username = 'test123',
ssh_pkey = self.pem,
remote_bind_address = ('test.com', 3306)
) as tunnel:
self.db = mysql.connector.MySQLConnection(**self.config, port = tunnel.local_bind_port)
def connection(self):
return self.db
После создания экземпляра с помощью
db = DB()
conn = db.connection()
Я пытаюсь выполнить следующий код:
cursor = conn.cursor()
query = ("""
SELECT *
FROM user
WHERE user_id = 1
""")
cursor.execute(query)
rows = cursor.fetchall()
print(rows)
И я получаю следующее сообщение:
Соединение MySQL недоступно.