Подключение к БД 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 шт):

Автор решения: CrazyElf

Вместо самописной библиотеки 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)
→ Ссылка