【Python】oracledbでOracleに接続して操作する方法（サンプルコード付き）

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

PythonでOracleデータベースへ接続する

python-oracledbとは？

python-oracledbは、以前広く使われていて cx_Oracle の後継として Oracle 社が提供する公式の Python ライブラリです。ライブラリ名はシンプルに「oracledb」です。これにより、Oracleデータベースに対して Python から簡単に接続・操作することができます。

大きな特徴は、Oracle Client が不要なThinモードに対応しており、手軽に導入ができる点です。一方で、従来の cx_Oracle 同様の機能を活用したい場合は、Thick モードに切り替えての使用も可能です。

oracledbのインストール方法

oracledb ライブラリは、PyPIで公開されており、以下のように pip install コマンドで簡単にインストールできます。

pip install oracledb

ThinモードとThickモードの違いと選び方

oracledbは、Thin モードと Thick モードの2つの動作モードを提供しています。

Thin モードは Oracle Client も不要でとても便利です。Thin モードで問題ない場合は、そのままで十分ですが Advanced Queuing などの Oracle 特有機能を使いたい場合には、Thick モードの使用を検討する必要があります。

oracledbを用いたサンプルプログラム

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

前提条件

サンプルコード

設定ファイルおよびDBアクセスクラス定義

oracledb を使用して Oracle データベースへアクセスするための設定ファイルとDBアクセス用クラスの例は以下のようになります。設定ファイルの内容はお使いの環境にあわせて修正してください。

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

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

【db_connect_oracle.py】DBアクセスクラス

import configparser

import oracledb


class DbConnectOracle:
    """Oracle DBアクセスクラス"""

    def __init__(self, mode: str = "thin", client_lib_dir: str = None) -> None:
        """コンストラクタ

        Args:
            mode: "thin" または "thick"
            client_lib_dir: Thick モードの場合は、Oracle Client のライブラリパス
        """
        # モード判定
        if mode == "thick":
            if client_lib_dir is None:
                raise ValueError(
                    "Thick モードでは client_lib_dir の指定が必要です。"
                )
            oracledb.init_oracle_client(client_lib_dir)
        elif mode != "thin":
            raise ValueError(
                "mode は、'thin' または 'Thick' を指定してください。"
            )

        # iniファイルの読み込み
        config_db = configparser.ConfigParser()
        config_db.read("dbconfig.ini")

        # 設定の読み込み
        host = config_db["ORACLE_DB_SERVER"]["host"]
        port = int(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 = oracledb.makedsn(host, port, service_name=service_name)

        # 接続の確立
        self.con = oracledb.connect(
            user=user,
            password=password,
            dsn=self.dsn,
        )
        self.cursor = self.con.cursor()

    def execute_non_query(self, sql: str, bind_var: dict = None) -> None:
        """SQL 実行メソッド
        (CREATE / INSERT / UPDATE / DELETE 用)

        Args:
            sql: 実行SQL
            bind_var: バインド変数

        Returns: 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:
        """SQL 実行メソッド
        (SELECT 用)

        Args:
            sql: 実行SQL
            bind_var: バインド変数
            count: データ取得件数

        Returns: 取得結果リスト

        """
        # SQLの実行
        if bind_var is None:
            self.cursor.execute(sql)
        else:
            # バインド変数がある場合は指定して実行
            self.cursor.execute(sql, bind_var)

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

        # 取得結果を返却 (件数指定がある場合は、指定された件数分返却)
        return (
            self.cursor.fetchall()
            if count == 0
            else self.cursor.fetchmany(count)
        )

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

        Returns: None

        """
        self.con.commit()

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

        Returns: None

        """
        self.con.rollback()

    def __del__(self) -> None:
        """デストラクタ

        Returns: None

        """
        self.cursor.close()
        self.con.close()


if __name__ == "__main__":
    pass

使用例（Thin モード）

oracledb の Thin モードを使用して上記で作成したクラスを用いてOracleへの接続をするには以下のように使用します。

from db_connect_oracle import DbConnectOracle


def main():
    # Thin モードでアクセスする場合
    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()

使用例（Thick モード）

oracledb の Thick モードを使用する場合には、クラスをインスタンス化する際に以下の引数を設定してください。

• mode："thick"
• client_lib_dir：Oracleクライアントライブラリが格納されているパス

以下の例では、Oracle の Instant Client (軽量版クライアント) を使用した場合の例です。フルのOracle Clientの場合は、環境にあわせて r"C:\oracle\product\19.0.0\client_1\bin" のような指定をしていただければいいかと思います。

from db_connect_oracle import DbConnectOracle


def main():
    mode = "thick"
    client_lib_dir = r"C:\oracle\instantclient_21_18"

    # Thick モードでアクセスする場合
    db_oracle = DbConnectOracle(
        mode=mode,
        client_lib_dir=client_lib_dir,
    )

    # 以降は、Thin モードと同じ
    # ・・・


if __name__ == "__main__":
    main()

サンプルコードのポイント解説

DBアクセス用クラス DBConnectOracle

DBアクセス用のクラスとしてDbConnectOracleというクラスを定義しています。メソッドとして定義しているものは以下になります。

まとめ

python-oracledbを用いてOracleデータベースへアクセスする方法について解説しました。

このライブラリは、Oracle公式であり、Oracleアクセス手段として非常に有力です。これまで使われていた cx_Oracle の後継として使用できます。

Thin モードと Thick モードが用意されており、Thin モードではOracle Client不要で手軽に使用することができます。また、高度な機能が必要であれば Thick モードの使用も検討できます。

この記事では、汎用的に使えるDBアクセスクラスを作成してOracleDBへのアクセスをできるようにしてみました。ぜひ、参考にしていただいてOracle接続に活用していただければと思います。

【Python Tech】Pythonプログラミングガイド本サイトで紹介しているPython入門、応用に関する記事のまとめページです。Python文法の入門から高度な内容、おすすめの書籍についても紹介しています。また、機械学習等のデータ分析やGUI開発、近年話題の生成AI等のトピックスについても含んでいます。...