Get a connection and execute a SQL in Oracle with python.

DoOracleSQL.py

import cx_Oracle
import os

# Data Base configuration
db_host = os.environ.get('DB_HOST')
db_port = os.environ.get('DB_PORT')
db_name = os.environ.get('DB_NAME')
db_user = os.environ.get('DB_USER')
db_pass = os.environ.get('DB_PASS')


class OracleConn:
    """
    Encapsulate functions needed to query a BD
    """
    conn = None
    dsn = None

    def genDSN(self):
        """
        Generate the connection string from environment variables.
        :return: just the connection string.
        """
        self.dsn = cx_Oracle.makedsn(
                host=db_host,
                port=db_port,
                sid=db_name
            )

    def genConn(self):
        """
        Retrieve a connection to the Data base instance.
        :return:
        """
        self.genDSN()
        self.conn = cx_Oracle.connect(
                user=db_user,
                password=db_pass,
                dsn=self.dsn
            )

    def read_query_file(self, sqlfile: str):
        f = open(sqlfile, "r")
        if f.mode == "r":
            return f.read()

    def executeO66(self, sql66: str, arguments: dict):
        """
        Execute the order (sql)66: Usually I use text based-Store Procedures (I strongly recomend do that), so you will
        need to provide some arguments.
        :param sql66: The name of the file that store the text based store procedure to execute.
        :param arguments: The arguments to feed the Store Procedure.
        :return:
        """
        rtn = []
        if not self.conn:
            self.genConn()
        try:
            cursor = self.conn.cursor()
            cursor.execute(
                    self.read_query_file(sql66),
                    arguments
                )
            res = cursor.fetchall()
            for row in res:
                # To-do: bidding to list? to models? into a lambda? I dunno
                rtn.append(row)
        except cx_Oracle.DatabaseError as e:
            print("There is a problem with Oracle", e)
        finally:
            if cursor:
                cursor.close()
            if self.conn:
                self.conn.close()
        return rtn

    def version(self):
        return self.conn.version

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