Подключение к БД Oracle из телеграм бота
есть телеграм бот:
import logging
from datetime import datetime
from oracle_db import OracleDBConnector
import telebot
import cx_Oracle
# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
token = TOKEN
bot = telebot.TeleBot(token)
def connect_to_db():
try:
logging.info("Attempting to connect to the database.")
db = OracleDBConnector('instantclient_11_2', '10.10.5.111', 1521, 'med', 'user', 'qwerty')
connection = db.connect()
logging.info("Successfully connected to the database.")
return connection
except cx_Oracle.Error as error:
logging.error("Error occurred during connection: %s", error)
return None
def fetch_data_from_db(connection):
cursor = None
data = None
try:
if connection:
cursor = connection.cursor()
query = "SELECT * FROM patient WHERE keyid = 1"
cursor.execute(query)
data = cursor.fetchall() # Получить все строки сразу
logging.info("Query executed successfully. Fetched rows: %s", data)
else:
logging.warning("No valid connection available.")
except cx_Oracle.Error as error:
logging.error("Error occurred during query execution: %s", error)
finally:
if cursor:
cursor.close()
logging.info("Cursor closed.")
return data
@bot.message_handler(func=lambda message: True)
def handle_text(message):
user_text = message.text
logging.info("Received message: %s", user_text)
if "test" in user_text.lower():
logging.info("Processing 'test' command.")
connection = connect_to_db()
if connection:
data = fetch_data_from_db(connection)
if data:
logging.info("Data fetched from database: %s", data)
bot.reply_to(message, f"Fetched data: {data}")
else:
logging.warning("No data fetched from the database.")
bot.reply_to(message, "No data fetched from the database.")
connection.close()
logging.info("Database connection closed.")
else:
logging.warning("Failed to establish connection.")
bot.reply_to(message, "Failed to establish connection to the database.")
bot.polling(none_stop=True)
Также есть класс, который устанавливает соединение с БД:
import cx_Oracle
class OracleDBConnector():
'''
Класс работы с БД
'''
def __init__(self, client, addr, port, service_name, schemaName, passwd):
self.schemaName = None
self.passwd = None
self.connection = None
self.cursor = None
self.addr = addr
self.port = port
self.service_name = service_name
self.schemaName = schemaName
self.passwd = passwd
try:
cx_Oracle.init_oracle_client(client)
except:
pass
def connect(self):
'''
Подключение к БД
:param addr: адрес
:param port: порт
:param service_name: имя
:param schemaName: пользователь/схема
:param passwd: пароль
:return:
'''
if self.connection:
self.connection.close()
self.connection = None
self.cursor = None
dns_tns = cx_Oracle.makedsn(self.addr, self.port, service_name=self.service_name)
try:
self.connection = cx_Oracle.connect(user=self.schemaName, password=self.passwd, dsn=dns_tns)
self.cursor = self.connection.cursor()
except:
pass
def query(self, queryText, exc=False):
try:
cur = self.connection.cursor()
cur.execute(queryText)
if cur.description:
return list(cur)
else:
return None
except:
if exc:
return
self.connect()
self.query(queryText, True)
def querydict(self, queryText, exc=False):
try:
cur = self.connection.cursor()
self.addQueryLog(queryText)
cur.execute(queryText)
res = []
if cur.description:
format_dict = {}
for f in cur.description:
format_dict[f[0]] = None
cnt = 0
for rec in list(cur):
r_dict = format_dict.copy()
for field in rec:
r_dict[list(format_dict.keys())[cnt]] = field
cnt += 1
res.append(r_dict)
cnt = 0
return res
else:
return None
except:
if exc:
return
self.connect()
self.querydict(queryText, True)
def commit(self):
self.connection.commit()
def rollback(self):
self.connection.rollback()
def closeConnect(self):
self.connection.close()
def addQueryLog(self, txt):
with open('query.txt', 'a+') as f:
f.write('Query:\n')
f.write(txt)
f.write('\n')
Когда пишу боту выходит ошибка:
2024-06-24 09:57:38,684 - INFO - Received message: test 2024-06-24 09:57:38,684 - INFO - Processing 'test' command. 2024-06-24 09:57:38,684 - INFO - Attempting to connect to the database. 2024-06-24 09:57:39,372 - INFO - Successfully connected to the database. 2024-06-24 09:57:39,372 - WARNING - Failed to establish connection.
Подскажите пожалуйста с чем может быть связана ошибка?
Настройки подключения к БД верные.
Ответы (1 шт):
Вместо самописной библиотеки oracle_db
, которая работает не совсем так, как вы думали (возвращает None
при вызове connect
) и не показывает исключения, а просто их глотает, лучше используйте библиотеку oracledb
, она же python-oracledb.
pip install oracledb
Пример использования:
import getpass
import oracledb
un = 'scott'
cs = 'localhost/orclpdb'
pw = getpass.getpass(f'Enter password for {un}@{cs}: ')
with oracledb.connect(user=un, password=pw, dsn=cs) as connection:
with connection.cursor() as cursor:
sql = """select sysdate from dual"""
for r in cursor.execute(sql):
print(r)