このドキュメントでは、Pythonのcx_Oracleドライバーを使用してOceanBaseデータベースのOracleモードに接続し、操作する方法について説明します。cx_OracleはOracleが公式に提供するPythonデータベースインターフェースであり、Oracle OCIと互換性があり、OceanBaseデータベースのOracleテナントにシームレスに接続できます。
前提条件
- Python 3.7以降のバージョンがインストールされていること。
- OceanBaseデータベースがインストール済みで、Oracleテナントが作成されていること。
- libobclientとOBCIコンポーネントがインストールされていること。
手順
以下の手順に従って、cx_Oracleドライバを使用してOceanBaseデータベースのOracleモードに接続します。
ステップ１：Python 3.7環境のインストール
システムにPython 3.7以降がインストールされていることを確認してください。以下のコマンドでPythonのバージョンを確認できます：
python3 --version
インストールされていない場合は、お使いのOSに応じてPython 3.7をインストールしてください。
ステップ2：libobclientとOBCIコンポーネントのインストール
OceanBaseクライアントコンポーネント、すなわちlibobclientとOBCIをインストールします：
# 古いバージョンをアンインストール（存在する場合）
rpm -e libobclient
rpm -e obci
# 新しいバージョンをインストール
rpm -ivh libobclient-2.2.11-42025062010.el7.x86_64.rpm
rpm -ivh obci-2.1.1-342025070917.el7.x86_64.rpm
注意
libobclientのバージョンは2.2.11以上、obciのバージョンは2.1.1以上であることを確認してください。バージョンが低すぎると接続失敗や機能異常が発生する可能性があります。
ステップ3：cx_Oracleドライバーのインストール
cx_Oracleドライバーを解凍してインストールします：
# ドライバーのインストールディレクトリに移動
cd /directory/of/cx_Oracle-8.3.0.tar.gz
# cx_Oracleドライバーパッケージを解凍
tar -xvf cx_Oracle-8.3.0.tar.gz
# 解凍したディレクトリに移動
cd cx_Oracle-8.3.0
# cx_Oracleドライバーをインストール
python3 setup.py install
インストール完了後、システムは /usr/local/lib/python3.7/site-packages/ ディレクトリにcx_Oracleモジュールを生成します。
ステップ4：環境変数の設定
LD_LIBRARY_PATH環境変数を設定し、PythonがOceanBaseクライアントライブラリを見つけられるようにします：
export LD_LIBRARY_PATH=/u01/obclient/lib/:$LD_LIBRARY_PATH
この設定は、永久に有効にするために、シェル設定ファイル（例：~/.bashrc または ~/.zshrc）に追加することを推奨します。
ステップ5：テストプログラムを作成する
test_oracle.py という名前のファイルを作成し、以下の内容を含めます：
import cx_Oracle
# データベース接続情報
username = 'test@oracle'
password = 'test'
oracle_connection = 'xxx.xxx.xxx.xxx:2881/TEST'
# データベース接続の作成
conn = cx_Oracle.connect(username, password, oracle_connection)
def exec_sql(sql):
"""SQLステートメントの実行"""
cur = conn.cursor()
try:
cur.execute(sql)
conn.commit()
print(f"SQL実行成功: {sql}")
except Exception as e:
print(f"SQL実行失敗: {sql}")
print(f"エラーメッセージ: {e}")
finally:
cur.close()
def print_data(sql):
"""データのクエリと出力"""
cur = conn.cursor()
try:
cur.execute(sql)
data = cur.fetchall()
print(f"クエリ結果: {data}")
return data
except Exception as e:
print(f"クエリ失敗: {sql}")
print(f"エラーメッセージ: {e}")
return None
finally:
cur.close()
def test_char_types():
"""文字型のテスト"""
print("\n=== 文字型のテスト ===")
exec_sql("DROP TABLE test_char")
exec_sql("""
CREATE TABLE test_char (
id INT,
a VARCHAR2(20),
b CHAR(10),
c NCHAR(10),
d NVARCHAR2(10)
)
""")
exec_sql("INSERT INTO test_char VALUES (1, 'hello', 'adffdf', '2df4d', 'dsf44f')")
print_data("SELECT * FROM test_char")
def test_number_types():
"""数値型のテスト"""
print("\n=== 数値型のテスト ===")
exec_sql("DROP TABLE test_number")
exec_sql("""
CREATE TABLE test_number (
a NUMBER,
b FLOAT(126),
c BINARY_FLOAT,
d BINARY_DOUBLE
)
""")
exec_sql("INSERT INTO test_number VALUES (12.32, 12.34, 14.23, 123.3433)")
print_data("SELECT * FROM test_number")
def test_time_types():
"""時刻型のテスト"""
print("\n=== 時刻型のテスト ===")
exec_sql("DROP TABLE test_time")
exec_sql("""
CREATE TABLE test_time (
a DATE,
b TIMESTAMP,
c TIMESTAMP WITH TIME ZONE,
d TIMESTAMP WITH LOCAL TIME ZONE
)
""")
exec_sql("""
INSERT INTO test_time VALUES (
TIMESTAMP'2022-08-29 14:44:30',
TIMESTAMP'2022-08-29 14:44:30',
TIMESTAMP'2022-08-29 14:44:30',
TIMESTAMP'2022-08-29 14:44:30'
)
""")
print_data("SELECT * FROM test_time")
def test_lob_types():
"""LOB型のテスト"""
print("\n=== LOB型のテスト ===")
exec_sql("DROP TABLE test_lob")
exec_sql("""
CREATE TABLE test_lob (
a CLOB,
b BLOB,
c RAW(100)
)
""")
exec_sql("INSERT INTO test_lob VALUES ('sdfdslkfjldsf', '31323334353637', '31323334')")
print_data("SELECT * FROM test_lob")
def main():
"""メイン関数"""
try:
print("OceanBaseデータベースのOracleモード接続テストを開始します...")
# 各種データ型のテスト
test_char_types()
test_number_types()
test_time_types()
test_lob_types()
print("\n全てのテストが完了しました！")
except Exception as e:
print(f"テスト中にエラーが発生しました: {e}")
finally:
# データベース接続を閉じる
if conn:
conn.close()
print("データベース接続が閉じられました")
if __name__ == "__main__":
main()
ステップ6：テストプログラムを実行する
コマンドラインでテストプログラムを実行します：
python3 test_oracle.py
プログラムは、文字型、数値型、時刻型、LOB型のデータ操作を順次テストし、実行結果を出力します。
接続プールのサポート
OBCI 2.1.1バージョンから、接続プール機能がサポートされています。接続プールを使用することで、アプリケーションのパフォーマンスと同時実行処理能力を向上させることができます。
import cx_Oracle
from cx_Oracle import SessionPool
def test_simple_pool():
"""接続プールテスト"""
try:
print("接続プールの作成...")
# 接続プールの作成
pool = SessionPool(
user="test@oracle",
password="test",
dsn="xxx.xxx.xxx.xxx:2881/TEST",
min=2,
max=5,
increment=1
)
print(f"接続プールの作成に成功しました！")
print(f"- 最小接続数: {pool.min}")
print(f"- 最大接続数: {pool.max}")
print(f"- 増分: {pool.increment}")
print("\n接続プールの使用をテストします...")
# 接続プールから接続を取得
with pool.acquire() as connection:
with connection.cursor() as cursor:
# クエリの実行
cursor.execute("SELECT COUNT(*) FROM test_char")
result = cursor.fetchone()
print(f"test_char テーブルのレコード数: {result[0]}")
# 別のクエリの実行
cursor.execute("SELECT * FROM test_char WHERE id = 1")
data = cursor.fetchone()
print(f"クエリ結果: {data}")
print(f"\n接続プールの状態:")
print(f"- 現在開いている接続数: {pool.opened}")
print(f"- 現在忙しい接続数: {pool.busy}")
print("\n接続プールテストが正常に完了しました！")
# 注意：pool.close() を呼び出さないことで、ダンプエラーを回避します
except Exception as e:
print(f"エラーが発生しました: {e}")
if __name__ == "__main__":
test_simple_pool()