【Python】cx_Oracleを用いたOracleデータベースへのアクセス方法

Python で cx_Oracle を用いてOracleデータベースへアクセスする方法について解説します。

Python で Oracle データベースへアクセスする

業務アプリケーションなどを開発する際には、データベースとして Oracle 社の Oracle データベースがよく使用されます。Python のプログラムから Oracle データベースへアクセスする方法として、cx_Oracle モジュールを使用できます。

cx_Oracle は Oracle 社が自ら作成しているもので、基本的な SQL アクセスやPL/SQL の実行などの機能を提供しています。

cx_Oracle のインストール

cx_Oracle を使用するには、以下のコマンドを用いて pip でインストールを行ってください。

pip install cx_Oracle

pip インストールが完了すれば、cx_Oracle を用いて Oracle データベースへアクセスするための準備ができます。以降では、cx_Oracle のデータベースへアクセスするためのクラスを作って、Oracle データベースへアクセスする方法を紹介します。

cx_Oracle を用いた Oracle データベースへのアクセス

以降では、cx_Oracle を使用して Oracle データベースへアクセスするためのプログラムを紹介していきます。

サンプルプログラムの前提条件

以降で紹介するサンプルプログラムを実行するにあたっての前提条件は以下とします。

今回は無償版の OracleXE を使用しますが、製品版の Oracle でも同様に使用できます。以降で設定ファイルの記載も含めて紹介していますので、各環境にあわせて設定ファイルの接続先やサービス名、または一部プログラムを変更してもらえば異なる環境で動かすことが可能です。上記前提は、コードをそのまま実行する場合の前提条件ということでご理解ください。

cx_Oracle を用いた Oracle データベースアクセスのためのサンプルプログラム

cx_Oracle を用いて Oracle データベースへアクセスするための設定ファイルおよびサンプルプログラム、実行結果は以下のようになります。

前述した前提に従った環境であれば、このままコピーしていただくだけでも使用することができるかと思います。それぞれの具体的な詳細については、次節で説明します。

【dbconfig.ini】DB アクセス設定ファイル

[ORACLE_DB_SERVER]
host = localhost
port = 1521
service = XEPDB1
user = TEST
password = PAssw0rd

【db_connect_oracle.py】DB アクセス用クラスと使い方例を含むプログラム

import cx_Oracle
import configparser


class DbConnectOracle:
    def __init__(self) -> None:
        """コンストラクタ
        :return: None
        """
        # コンフィグファイルからデータを取得
        config_db = configparser.ConfigParser()
        config_db.read("dbconfig.ini")

        # 接続情報の取得
        host = config_db["ORACLE_DB_SERVER"]["host"]
        port = config_db["ORACLE_DB_SERVER"]["port"]
        service_name = config_db["ORACLE_DB_SERVER"]["service"]
        user = config_db["ORACLE_DB_SERVER"]["user"]
        password = config_db["ORACLE_DB_SERVER"]["password"]

        # DSN(データソース名)を作成する
        self.dsn = cx_Oracle.makedsn(host, port, service_name=service_name)

        # コネクションを確立する
        self.con = cx_Oracle.connect(
            user=user, password=password, dsn=self.dsn, encoding="UTF-8"
        )

        # カーソルを作成する
        self.cursor = self.con.cursor()

    def execute_non_query(self, sql: str, bind_var: dict = None) -> None:
        """CREATE/INSERT/UPDATE/DELETEのSQL実行メソッド
        :param sql: 実行SQL
        :param bind_var: バインド変数
        :return: None
        """
        # SQLの実行
        if bind_var is None:
            self.cursor.execute(sql)
        else:
            # バインド変数がある場合は指定して実行
            self.cursor.execute(sql, bind_var)

    def execute_query(self, sql: str, bind_var: dict = None, count: int = 0) -> list:
        """SELECTのSQL実行メソッド
        :param sql: 実行SQL
        :param bind_var: バインド変数
        :param count: データ取得件数
        :return: 結果リスト
        """
        # SQLの実行
        if bind_var is None:
            self.cursor.execute(sql)
        else:
            # バインド変数がある場合は指定して実行
            self.cursor.execute(sql, bind_var)

        result = []
        if count == 0:
            columns = [col[0] for col in self.cursor.description]
            self.cursor.rowfactory = lambda *args: dict(zip(columns, args))
            rows = self.cursor.fetchall()
            for row in rows:
                result.append(row)
        else:
            # 件数指定がある場合はその件数分を取得する
            columns = [col[0] for col in self.cursor.description]
            self.cursor.rowfactory = lambda *args: dict(zip(columns, args))
            rows = self.cursor.fetchmany(count)
            for row in rows:
                result.append(row)

        return result

    def commit(self) -> None:
        """コミット
        :return: None
        """
        self.con.commit()

    def rollback(self) -> None:
        """ロールバック
        :return: None
        """
        self.con.rollback()

    def __del__(self) -> None:
        """デストラクタ
        :return: None
        """
        self.cursor.close()
        self.con.close()


def main():
    db_oracle = DbConnectOracle()

    # ===== テーブルを作成する
    create_sql = (
        "CREATE TABLE SAMPLE_TABLE"
        "("
        "ID NUMBER GENERATED ALWAYS AS IDENTITY,"
        "STR1 VARCHAR2(50), "
        "VALUE1 NUMBER, "
        "LAST_UPDATE_DATETIME TIMESTAMP, "
        "PRIMARY KEY (ID)"
        ")"
    )
    db_oracle.execute_non_query(create_sql)
    # コミット
    db_oracle.commit()

    # ===== データをINSERTする
    insert_sql = (
        "INSERT INTO SAMPLE_TABLE "
        "(STR1, VALUE1, LAST_UPDATE_DATETIME) "
        "VALUES(:str1, :val1, CURRENT_TIMESTAMP)"
    )
    for i in range(5):
        input_str = f"test_str{i}"
        input_val = 10 * i
        bind = {"str1": input_str, "val1": input_val}
        db_oracle.execute_non_query(insert_sql, bind)
    # コミット
    db_oracle.commit()

    # ===== データを検索する (全て検索)
    print("===== 全件検索")
    select_sql = "SELECT * FROM SAMPLE_TABLE"
    select_result = db_oracle.execute_query(select_sql)
    print(select_result)

    # ===== 件数を指定して検索する (以下の例は3件)
    print("===== 取得する件数の指定")
    select_sql = "SELECT * FROM SAMPLE_TABLE"
    select_result = db_oracle.execute_query(select_sql, count=3)
    print(select_result)

    # ===== 条件を指定して検索する
    print("===== 条件を指定して実行する")
    select_sql = "SELECT * FROM SAMPLE_TABLE WHERE VALUE1 > :val1"
    print("条件１")
    select_result = db_oracle.execute_query(select_sql, {"val1": 10})
    print(select_result)
    print("条件２")
    select_result = db_oracle.execute_query(select_sql, {"val1": 30})
    print(select_result)


if __name__ == "__main__":
    main()

【実行結果】
===== 全件検索
[{'ID': 1, 'STR1': 'test_str0', 'VALUE1': 0, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 2, 'STR1': 'test_str1', 'VALUE1': 10, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 3, 'STR1': 'test_str2', 'VALUE1': 20, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 4, 'STR1': 'test_str3', 'VALUE1': 30, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 5, 'STR1': 'test_str4', 'VALUE1': 40, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}]
===== 取得する件数の指定
[{'ID': 1, 'STR1': 'test_str0', 'VALUE1': 0, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 2, 'STR1': 'test_str1', 'VALUE1': 10, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 3, 'STR1': 'test_str2', 'VALUE1': 20, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}]
===== 条件を指定して実行する
条件１
[{'ID': 3, 'STR1': 'test_str2', 'VALUE1': 20, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 4, 'STR1': 'test_str3', 'VALUE1': 30, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}, {'ID': 5, 'STR1': 'test_str4', 'VALUE1': 40, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}]
条件２
[{'ID': 5, 'STR1': 'test_str4', 'VALUE1': 40, 'LAST_UPDATE_DATETIME': datetime.datetime(2022, 4, 15, 6, 15, 26, 318000)}]

サンプルコードの詳細説明

DB接続設定ファイル dbconfig.ini

まずは、DB 接続設定ファイルを dbconfig.ini として用意します。ファイル名は呼び出し時の指定と揃えていただければ、任意につけていただいて構いません。以降で説明する DB アクセスクラスのプログラム側では configparser を用いて、この設定ファイルの内容を取得して使用します。

【dbconfig.ini】DB アクセス設定ファイル（再掲）

[ORACLE_DB_SERVER]
host = localhost
port = 1521
service = XEPDB1
user = TEST
password = PAssw0rd

具体的には、host (ホスト)、port (ポート)、service (サービス名)、user (ユーザー名)、password (パスワード)を記載しています。

host については、具体的なサーバーのホスト名や IP アドレスでも構いません。今回はローカル PC の DB を使うため localhost としています。port については、Oracleでデフォルトで使用されるポートです。service、user、password は環境にあわせて指定します。

設定ファイルにせずに上記情報をプログラムに直接記載しても問題ないのですが、メンテナンスしづらいので設定ファイルとして切り出しています。また、パスワードをそのまま平文で記載しておくのはよくありませんが、今回は簡単にするため、そのままとしています。

DB アクセス用クラス DBConnectOracle

DB アクセス用のクラスとして DbConnectOracle というクラスを定義しています。

メソッドとして定義しているものは以下になります。

それぞれのメソッドのポイントとなる内容について以降で解説していきます。

コンストラクタ：__init__

コンストラクタでは、DB への接続を確立します。まずは、以下の部分でコンフィグファイルから設定を取得してきます。

    def __init__(self) -> None:
        """コンストラクタ
        :return: None
        """
        # コンフィグファイルからデータを取得
        config_db = configparser.ConfigParser()
        config_db.read('dbconfig.ini')

        # 接続情報の取得
        host = config_db['ORACLE_DB_SERVER']['host']
        port = config_db['ORACLE_DB_SERVER']['port']
        service_name = config_db['ORACLE_DB_SERVER']['service']
        user = config_db['ORACLE_DB_SERVER']['user']
        password = config_db['ORACLE_DB_SERVER']['password']

上記部分により、host などの情報を設定ファイルから取得できます。その後、以下の部分で接続のためのコネクション確立と DB 操作のためのカーソルを用意しています。

        # DSN(データソース名)を作成する
        self.dsn = cx_Oracle.makedsn(host, port, service_name=service_name)

        # コネクションを確立する
        self.con = cx_Oracle.connect(
            user=user,
            password=password,
            dsn=self.dsn,
            encoding='UTF-8'
        )

        # カーソルを作成する
        self.cursor = self.con.cursor()

DSN (データソース名)は、makedsn メソッドで組み立てることができます。self.dsn を print してみてもらえれば分かりますが、以下のような接続用の文字列を組みたてることができます。

(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=localhost)(PORT=1521))(CONNECT_DATA=(SERVICE_NAME=XEPDB1)))

connect メソッドに user、password、dsn、encoding を指定することでコネクションが確立できます。また、カーソルはコネクションの cursor メソッドで作成できます。

SQL 実行：execute_non_query

SQL 実行用のメソッドとして execute_non_query を用意しています。後述するexecute_query との違いは返却値がないことで、CREATE / INSERT / UPDATE / DELETE などの SQL を実行する用として作成しています。

    def execute_non_query(self, sql: str, bind_var: dict = None) -> None:
        """CREATE/INSERT/UPDATE/DELETEのSQL実行メソッド
        :param sql: 実行SQL
        :param bind_var: バインド変数
        :return: None
        """
        # SQLの実行
        if bind_var is None:
            self.cursor.execute(sql)
        else:
            # バインド変数がある場合は指定して実行
            self.cursor.execute(sql, bind_var)

SQL を実行するためには、カーソルの execute メソッドに対象となる SQL を渡します。また、バインド変数がある場合は、引数に辞書を渡して実行できます。

SQL 実行：execute_query

SQL 実行用のメソッドとしてもう一つ execute_query を用意しています。上記の execute_non_query との違いは返却値があることで SELECT してデータを取得する用として作成しています。

    def execute_query(self,
                      sql: str,
                      bind_var: dict = None,
                      count: int = 0) -> list:
        """SELECTのSQL実行メソッド
        :param sql: 実行SQL
        :param bind_var: バインド変数
        :param count: データ取得件数
        :return: 結果リスト
        """
        # SQLの実行
        if bind_var is None:
            self.cursor.execute(sql)
        else:
            # バインド変数がある場合は指定して実行
            self.cursor.execute(sql, bind_var)

        result = []
        if count == 0:
            columns = [col[0] for col in self.cursor.description]
            self.cursor.rowfactory = lambda *args: dict(zip(columns, args))
            rows = self.cursor.fetchall()
            for row in rows:
                result.append(row)
        else:
            # 件数指定がある場合はその件数分を取得する
            columns = [col[0] for col in self.cursor.description]
            self.cursor.rowfactory = lambda *args: dict(zip(columns, args))
            rows = self.cursor.fetchmany(count)
            for row in rows:
                result.append(row)

        return result

SQL を実行するためには、カーソルの execute メソッドに対象となる SQL を渡します。また、バインド変数がある場合には、引数に辞書を渡して実行できます。

SELECT 結果の全てを取得する場合は、fetchall メソッドで取得することができます。また、件数を指定する場合には、fetchmany メソッドで取得件数を指定すると指定件数分のみ取得できます。また、今回は使用していませんが 1 件取得する fetchone メソッドもあります。

cx_Oracle は返却値は、結果セットをタプルのリストで返します。しかし、辞書の方が列名が分かり扱いやすい場合があるので、以下の部分で辞書で返却するような設定にしています。

            columns = [col[0] for col in self.cursor.description]
            self.cursor.rowfactory = lambda *args: dict(zip(columns, args))

columns の部分ではカーソルの description メソッドで列名を取得しています。その後、カーソルの rowfactory に lamda 式として列名と値の辞書に変換する無名関数を設定しています。これにより SELECT の結果を辞書に変換してくれます。なお、上記の部分をコメントアウトすれば、結果はタプルのリストになります。

コミット：commit

データベースへの変更を確定させるためにはコミットをする必要があります。そのためのメソッドとして commit メソッドを用意しています。

    def commit(self) -> None:
        """コミット
        :return: None
        """
        self.con.commit()

コネクションには commit メソッドが用意されているため、実行することでコミットすることが可能です。

ロールバック：rollback

コミット前のデータベースへの変更を取り消したい場合は、ロールバックしてデータベースの状態を元に戻す必要があります。そのためのメソッドとして rollback メソッドが用意されています。

    def rollback(self) -> None:
        """ロールバック
        :return: None
        """
        self.con.rollback()

コネクションには rollback メソッドが用意されているため、実行することでロールバックすることが可能です。

デストラクタ：__del__

クラスのインスタンスが使用されなくなったら、DB アクセスのために用意していたカーソルとコネクションを閉じる必要があります。そのため、デストラクタである __del__ 内でクローズ処理を実施しています。

    def __del__(self) -> None:
        """デストラクタ
        :return: None
        """
        self.cursor.close()
        self.con.close()

カーソル及びコネクションには close メソッドが用意されているため、実行することでクローズ処理をすることができます。

DB アクセス用クラス DbConnectOracle の使い方
~テーブル作成、データ追加、データ検索例~

main 関数では、上記で定義した DbConnectOracle クラスを使用して、テーブルの作成 (CREATE TABLE)、データの追加 (INSERT)、データ検索 (SELECT)の使い方の例を記載しています。それぞれのパートの内容を順に説明します。

インスタンスの生成

まずは、以下のように作成した DB アクセスクラスをインスタンス化します。

    db_oracle = DbConnectOracle()

テーブルの作成（CREATE TABLE）

CREATE TABLE でテーブルを作成しているのが以下の部分です。

    # ===== テーブルを作成する
    create_sql = "CREATE TABLE SAMPLE_TABLE" \
                 "(" \
                 "ID NUMBER GENERATED ALWAYS AS IDENTITY," \
                 "STR1 VARCHAR2(50), " \
                 "VALUE1 NUMBER, " \
                 "LAST_UPDATE_DATETIME TIMESTAMP, " \
                 "PRIMARY KEY (ID)" \
                 ")"
    db_oracle.execute_non_query(create_sql)
    # コミット
    db_oracle.commit()

上記のように CREATE 用のSQLを用意して、execute_non_query メソッドに渡します。処理を確定する際には commit メソッドを呼び出します。

テーブルの存在有無確認などは含めていない SQL なので、複数回実行すると既にテーブルがあるためエラーとなります。その場合は、「テーブルを DROP する」又は「CREATE TABLE 部分のプログラムをコメントアウトする」などしてみてください。

データの登録（INSERT）

INSERT でデータを登録しているのが以下の部分です。

    # ===== データをINSERTする
    insert_sql = "INSERT INTO SAMPLE_TABLE " \
                 "(STR1, VALUE1, LAST_UPDATE_DATETIME) " \
                 "VALUES(:str1, :val1, CURRENT_TIMESTAMP)"
    for i in range(5):
        input_str = f'test_str{i}'
        input_val = 10 * i
        bind = {'str1': input_str, 'val1': input_val}
        db_oracle.execute_non_query(insert_sql, bind)
    # コミット
    db_oracle.commit()

今回の例では、5 件のデータを追加しています。SQL には値を埋め込む部分に「:str1」、 「:val1」といったバインド変数が用意してあり、execute_non_query メソッドにて辞書形式でバインド変数と値を指定することで INSERT を実行しています。処理を確定する際には、commit メソッドを呼び出します。

なお、今回 ID は「GENERATED ALWAYS AS IDENTITY」という指定をしているので、自動で採番がされます。

データの検索（SELECT）

SELECT で登録したデータを検索しているのが以下の部分です。今回は 3 つほど例を用意しています。

    # ===== データを検索する (全て検索)
    print('===== 全件検索')
    select_sql = "SELECT * FROM SAMPLE_TABLE"
    select_result = db_oracle.execute_query(select_sql)
    print(select_result)

    # ===== 件数を指定して検索する (以下の例は3件)
    print('===== 取得する件数の指定')
    select_sql = "SELECT * FROM SAMPLE_TABLE"
    select_result = db_oracle.execute_query(select_sql, count=3)
    print(select_result)

    # ===== 条件を指定して検索する
    print('===== 条件を指定して実行する')
    select_sql = "SELECT * FROM SAMPLE_TABLE WHERE VALUE1 > :val1"
    print('条件１')
    select_result = db_oracle.execute_query(select_sql, {'val1': 10})
    print(select_result)
    print('条件２')
    select_result = db_oracle.execute_query(select_sql, {'val1': 30})
    print(select_result)

全件検索する場合には「SELECT *」で検索すれば情報を取得できます。また、count 引数に数字を指定すれば全件のうち指定した件数のみ取得できます。

条件を指定する場合には、WHERE 句のある SQL をそのまま渡してもいいですが、上記の例では条件の値を「:val1」としてバインド変数で値を変えながら実行しています。これにより条件に一致するデータのみ検索出来ていることが分かるかと思います。

以上が、今回作成した DB アクセスクラスの簡単な使い方です。cx_Oracle を用いたOracle データベースへのアクセス方法として参考にしていただければと思います。

あわせて読みたい

【Python Tech】プログラミングガイド

ホッシー

システムエンジニア

Contact

はじめまして。当サイトをご覧いただきありがとうございます。 私は製造業のメーカーで、DX推進や業務システムの設計・開発・導入を担当しているシステムエンジニアです。これまでに転職も経験しており、以前は大手電機メーカーでシステム開発に携わっていました。

プログラミング言語はこれまでC、C++、JAVA等を扱ってきましたが、最近では特に機械学習等の分析でも注目されているPythonについてとても興味をもって取り組んでいます。これまでの経験をもとに、Pythonに興味を持つ方のお役に立てるような情報を発信していきたいと思います。どうぞよろしくお願いいたします。

※キャラクターデザイン：ゼイルン様