Как выполнять выражения и процедуры PL/SQL в Python

В этом руководстве мы будем использовать cx_Oracle — модуль расширения в Python, который включает доступ к базе данных Oracle.

Сценарий использования: здесь мы расскажем о том, как выполнять процедуры и выражения PL/SQL в Python. В данном примере показано, как выполнить процедуру, которая обновляет таблицу (напр. table1), а также как запросить эту таблицу из Python и конвертировать ее в датафрейм pandas.

Сначала создадим главный класс под названием Oracle(), в котором есть многочисленные функции, как показано ниже.

connect_node(). Данная функция начнет взаимодействие с базой Oracle, передав детали базы данных, такие как user, password, hostname, port и servicename. Мы будем использовать try и except для обработки исключений и документирования ошибок.
disconnect_node(). Эта функция запустит отключение от базы данных Oracle, закрывая базу и cursor. Если соединение отсутствует, то функция проигнорирует исключение.
execute_node(). Выполняет любые выражения SQL, переданные в функцию commit, если это указано. Тут мы не будем выделять fetchall(), так как выражение SQL может не быть select. Лучше позаботиться об этом отдельно при создании датафрейма из таблицы SQL.
execute_proc_node(). Выполняет любую процедуру SQL, переданную в функцию commit, если это указано.

import cx_Oracle
import pandas as pd
import json
conf = json.loads(open('conf.json').read())
password_key=list(conf['dict'].keys())[0]
password=conf['dict'][password_key]

class Oracle():
    def __init__(self):
        print('Database Connection')
        
    def connect_node(self, username=user, password=password, hostname=hostname, port=port, servicename=servicename):
        """ Connect to the database. """
        try:
            self.db = cx_Oracle.connect(username, password
                                , hostname + ':' + port + '/' + servicename)
        except cx_Oracle.DatabaseError as e:
            # При необходимости регистрирует ошибку
            raise
        # Если соединение с базой данных пройдет успешно, создайте курсор,
        # который мы будем использовать.
        self.cursor = self.db.cursor()

    def disconnect_node(self):
        """
        Disconnect from the database. If this fails, for instance
        if the connection instance doesn't exist, ignore the exception.
        """
        try:
            self.cursor.close()
            self.db.close()
        except cx_Oracle.DatabaseError:
            pass

    def execute_node(self, sql, commit=False):
        """
        Execute whatever SQL statements are passed to the method;
        commit if specified. Do not specify fetchall() in here as
        the SQL statement may not be a select.
        """
        try:
            self.cursor.execute(sql)
        except cx_Oracle.DatabaseError as e:
            # При необходимости регистрирует ошибку
            raise
        # Commit выполняется только при необходимости
        if commit:
            self.db.commit()

    def execute_proc_node(self, sql, commit=False):
        """
        Execute whatever SQL procedure are passed to the method;
        commit if specified.
        """
        try:
            self.cursor.callproc(sql)
        except cx_Oracle.DatabaseError as e:
            # При необходимости регистрирует ошибку
            raise
        # Commit выполняется только при необходимости
        if commit:
            self.db.commit()

Импортируем модуль oracle_connection. Необходимо выполнить функцию из модуля под названием execute_proc_node и передать название процедуры, которое сохранится в переменной query. В этой процедуре мы обновляем таблицу (напр. table1). Таким образом, передадим true для commit, чтобы сохранить данные после отключения от базы данных. Наконец, отсоединимся от базы данных с помощью функции disconnect_node.

import oracle_connection as oracle_connect

query = 'procedure_name'
# Парсинг данных
orc = oracle_connect.Oracle()
orc.connect_node()
try:
    orc.execute_proc_node(sql=query, commit=True)
finally:
    orc.disconnect_node()

Покажем другой пример в том же документе Python. Выполним выражение SQL и извлечем (fetchall()) сырые данные, а затем конвертируем все в датафрейм pandas.Также импортируем модуль oracle_connection. Выполняем метод из модуля под названием execute_node и передаем выражение SQL, которое сохраняется в запросе переменной. Сохраняем сырые данные в переменную data и проходим циклом через все столбцы. Сохраняем все это в список col_names. Далее необходимо конвертировать данные в датафрейм pandas и отсоединиться от базы данных. В конце зададим столбцам названия, передав список col_names в dataset.columns и переименовав столбцы.

import oracle_connection as oracle_connect
import pandas as pd

query = 'select * from schema.table_name'
# Парсинг данных
orc = Oracle()
orc.connect_node()

try:
    orc.execute_node(sql=query, commit=False)
    data = orc.cursor.fetchall()
    col_names = []
    for i in range(0, len(orc.cursor.description)):
        col_names.append(orc.cursor.description[i][0])
    dataset = pd.DataFrame(data)
finally:
    orc.disconnect_node()

dataset.columns = col_names
dataset.head()

Надеемся, что статья была для вас полезной. Спасибо за прочтение!

Как выполнять выражения и процедуры PL/SQL в Python

Report Page