ApsaraDB for OceanBase:OceanBase Database への PyMySQL を使用した接続

python-pymysql サンプルプロジェクトをダウンロードする

PS C:\Windows\system32> python --version
Python 3.7.0
PS C:\Windows\system32> pip --version
pip 22.3.1 from d:\python\python37\lib\site-packages\pip (python 3.7)

obclient -h$host -P$port -u$user_name -p$password -D$database_name

obclient -hxxx.xxx.xxx.xxx -P3306 -utest_user001 -p****** -Dtest

プロジェクトコードの概要

[クリック] python-pymysql to download the project code, which is a compressed file named プロジェクトコード（圧縮ファイル）をダウンロードするにはpython-pymysql.zip。

解凍すると、 という名前のフォルダーがあります。ディレクトリ構造は次のとおりです。python-pymysql。ディレクトリ構造は次のとおりです。

python-pymysql
├── config.py
├── test_sql.py
├── main.py
└── requirements.txt

ファイルとディレクトリの詳細：

• config.py：データベース接続構成を格納します。

• test_sql.py：SQL 文を格納します。

• main.py：プログラムのエントリポイントであり、テーブルの作成、データの挿入、データのクエリ、テーブルの削除などの基本的なデータベース操作のためのシンプルな CLI を提供します。

• requirements.txt：プロジェクトに必要な Python ライブラリとそのバージョン要件をリストします。

  説明

  コマンドを実行すると、必要なライブラリが自動的にインストールされます。このトピックで提供されているコードは、PyMySQL ライブラリのバージョン要件のみをリストしています。sudo pip install -r requirements.txt必要なライブラリを自動的にインストールするコマンド。

config.py のコード

ファイルのコードconfig.py ファイル

データベース接続情報を定義します。情報には次の項目が含まれます。

データベース接続情報を定義します。情報には、OceanBase Database に接続するための IP アドレス、ポート番号、ユーザー名、パスワード、データベース名、および文字セットが含まれます。

サンプルコード：

DB_CONFIG = {
    'host': '$host',
    'port': $port,
    'user': '$user_name',
    'password': '$password',
    'database': '$database_name',
    'charset': 'utf8mb4'
}

パラメーターの説明：

• host: OceanBase Database に接続するための IP アドレスです。

• port：OceanBase Database に接続するためのポート。デフォルトポートは 3306 です。

• user：テナントアカウントのユーザー名。

• password：テナントアカウントのパスワード。

• database：接続するデータベースの名前。

• charset：データベースに接続するための文字セット。

test_sql.py のコード

この Topic で説明されている test_sql.py ファイル内のコードは、テーブルの作成、データの挿入、データのクエリ、テーブルの削除などのデータベース操作のための SQL 文を定義しています。これらの SQL 文は、アプリケーションが PyMySQL を介してデータベースに接続した後に実行できます。

test_sql.py ファイルを構成するには、次の手順を実行します。

1. テーブルを作成するための SQL 文を定義します。

   3 つのフィールド id、name、age を持つ test_pymysql テーブルを作成する SQL 文を定義します。ここで、id は自動インクリメントのプライマリキーです。

   サンプルコード：

   CREATE_TABLE_SQL = '''
   CREATE TABLE test_pymysql (
   id INT(11) NOT NULL AUTO_INCREMENT,
   name VARCHAR(128) NOT NULL,
   age INT(11) NOT NULL,
   PRIMARY KEY (id)
   )
   '''

2. データを挿入するための SQL 文を定義します。

   テーブルに 3 つの行を挿入する SQL 文を定義します。各行には、 フィールドと フィールドが含まれます。各フィールドの値は、SQL 文の実行時にプレースホルダー として渡されます。test_pymysql テーブルで、それぞれ 名前 および 年齢 フィールド。各フィールドの値はプレースホルダーとして渡されます %sSQL ステートメントが実行されたとき。

   サンプルコード：

   INSERT_DATA_SQL = '''
   INSERT INTO test_pymysql (name, age) VALUES 
   (%s, %s),
   (%s, %s),
   (%s, %s)
   '''

3. データをクエリするための SQL 文を定義します。

   テーブルからすべてのデータをクエリする SQL 文を定義します。test_pymysql テーブル。

   サンプルコード：

   SELECT_DATA_SQL = '''
   SELECT * FROM test_pymysql
   '''

4. テーブルを削除するための SQL 文を定義します。

   削除する SQL ステートメントを定義します。test_pymysql テーブル。

   サンプルコード：

   DROP_TABLE_SQL = '''
   DROP TABLE test_pymysql
   '''

main.py のコード

このトピックで取得した main.py ファイルのコードは、pymysql モジュールを呼び出して OceanBase Database の MySQL モードに接続し、logging モジュールを呼び出してログを出力し、テーブルの作成、データの挿入、データのクエリ、テーブルの削除などのデータベース操作を実行します。

main.py ファイルを構成するには、次の手順を実行します。 ファイルで、次の手順を実行します。

1. 必要なモジュールをインポートします。

   1. インポート ログ記録 モジュール。

   2. インポート pymysql モジュール。

   3. インポート config.py モジュール。データベース接続情報を定義します。

   4. インポート test_sql.py モジュール。データベース操作用の SQL 文を定義します。

   サンプルコードは以下のとおりです。

   import logging
   import pymysql
   from config import DB_CONFIG
   from test_sql import CREATE_TABLE_SQL, INSERT_DATA_SQL, SELECT_DATA_SQL, DROP_TABLE_SQL # CREATE_TABLE_SQL、INSERT_DATA_SQL、SELECT_DATA_SQL、および DROP_TABLE_SQL を test_sql からインポート

2. ログのレベルと形式を設定し、スクリプトの実行開始を示す INFO ログを生成します。

   サンプルコードは以下のとおりです:

   logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
   logging.info('Start executing the script') # スクリプトの実行を開始

3. テーブルを作成するための関数を定義します。

   という名前の関数を定義します create_table() と データベース接続とカーソルオブジェクトのライフサイクルを管理する文。これにより、データベース接続とカーソルオブジェクトをメモリリークなしで安全に閉じることができます。テーブルを作成するための SQL 文を実行し、トランザクションをコミットして成功ログを生成するか、トランザクションをロールバックしてエラーログを生成します。

   サンプルコードは以下のとおりです。

   def create_table():
       logging.info('Start creating the table') # テーブルの作成開始
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   cursor.execute(CREATE_TABLE_SQL)
                   conn.commit()
                   logging.info('Table creation successful') # テーブルの作成成功
               except Exception as e:
                   conn.rollback()
                   logging.error('Table creation failed, Reason: %s' % e) # テーブルの作成失敗
   
   

4. データを挿入する関数を定義します。

   ロギングのレベルと形式を設定し、スクリプトの実行開始を示す INFO ログを生成します。insert_data() と データベース接続とカーソルオブジェクトのライフサイクルを管理する文。これにより、データベース接続とカーソルオブジェクトをメモリリークなしで安全に閉じることができます。データを挿入するための SQL 文を実行し、トランザクションをコミットして成功ログを生成するか、トランザクションをロールバックしてエラーログを生成します。

   サンプルコード：

   def insert_data():
       logging.info('データの挿入を開始します') # データの挿入を開始します
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   data = [('John', 20), ('Lucy', 25), ('Tom', 30)]
                   flattened_data = [d for item in data for d in item]
                   cursor.executemany(INSERT_DATA_SQL, [flattened_data])
                   conn.commit()
                   logging.info('データの挿入に成功しました') # データの挿入に成功しました
               except Exception as e:
                   conn.rollback()
                   logging.error('データの挿入に失敗しました。理由: %s' % e) # データの挿入に失敗しました。理由: %s
   

5. テーブルを作成する関数を定義します。

   という名前の関数を定義しますselect_data() でデータベースからデータをクエリします。 を使用 文を使用して、データベース接続とカーソルオブジェクトのライフサイクルを管理します。 これにより、データベース接続とカーソルオブジェクトをメモリリークなしで安全に閉じることができます。 実行する() SQL 文を実行するメソッドSELECT_DATA_SQL でデータをクエリします。 [fetchall()メソッドを使用してクエリ結果を取得し、結果を 1 行ずつ出力します。対象 ループ。

   サンプルコード：

   def select_data():
       # DB に接続し、データを選択します。
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               # SQL を実行します。
               cursor.execute(SELECT_DATA_SQL)
               # 結果を取得します。
               result = cursor.fetchall()
               # 結果をループ処理します。
               for row in result:
                   print(row)

6. データを挿入する関数を定義します。

   を定義しますdrop_table()定義済みのデータベース接続情報を使用する関数 (DB_CONFIG) とテーブルを削除するための SQL ステートメント (DROP_TABLE_SQL)。この関数はテーブルを削除し、操作結果を示すログを出力します。テーブルの削除に失敗した場合、エラーメッセージが出力されます。

   サンプルコード：

   def drop_table():
       logging.info('テーブルの削除を開始します') # テーブルの削除を開始します
       with pymysql.connect(**DB_CONFIG) as conn:
           with conn.cursor() as cursor:
               try:
                   cursor.execute(DROP_TABLE_SQL)
                   conn.commit()
                   logging.info('テーブルが正常に削除されました') # テーブルが正常に削除されました
               except Exception as e:
                   conn.rollback()
                   logging.error('テーブルの削除に失敗しました。理由: %s' % e) # テーブルの削除に失敗しました。理由: %s
   

7. プログラムのエントリを定義します。これは主にデータベース操作を実行するために使用されます。

   最初に、現在のモジュールがメインプログラムとして実行されているかどうかを確認します。該当する場合は、次の操作を実行します。

   1. 呼び出す create_table() テーブルを作成する関数。

   2. 呼び出す insert_data()テーブルにデータを挿入する関数。

   3. 呼び出す select_data()テーブルからデータをクエリする関数。

   4. 呼び出す drop_table() テーブルを削除する関数。

   サンプルコードは次のとおりです:

   if __name__ == '__main__':
       create_table()
       insert_data()
       select_data()
       drop_table()

8. スクリプトの実行が完了したことを示す INFO ログを生成します。

   サンプルコードは以下のとおりです。

   logging.info('Script execution completed')

完全なコード例

config.py

# データベース接続
DB_CONFIG = {
    'host': '$host',
    'port': $port,
    'user': '$user_name',
    'password': '$password',
    'database': '$database_name',
    'charset': 'utf8mb4'
}

test_sql.py

テーブルを作成する
CREATE_TABLE_SQL = '''
CREATE TABLE test_pymysql (
  id INT(11) NOT NULL AUTO_INCREMENT,
  name VARCHAR(128) NOT NULL,
  age INT(11) NOT NULL,
  PRIMARY KEY (id)
  )
'''

データを挿入する
INSERT_DATA_SQL = '''
INSERT INTO test_pymysql (name, age) VALUES 
(%s, %s),
(%s, %s),
(%s, %s)
'''

データをクエリする
SELECT_DATA_SQL = '''
SELECT * FROM test_pymysql
'''

テーブルを削除する
DROP_TABLE_SQL = '''
DROP TABLE test_pymysql
'''

main.py

import logging
import pymysql
from config import DB_CONFIG
from test_sql import CREATE_TABLE_SQL, INSERT_DATA_SQL, SELECT_DATA_SQL, DROP_TABLE_SQL

# logging設定を初期化
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# スクリプトの実行開始をログ出力
logging.info('Start executing the script')

# テーブル作成
def create_table():
    # テーブル作成開始をログ出力
    logging.info('Start creating the table')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                cursor.execute(CREATE_TABLE_SQL)
                conn.commit()
                # テーブル作成成功をログ出力
                logging.info('Table creation successful')
            except Exception as e:
                conn.rollback()
                # テーブル作成失敗をログ出力、理由と共に
                logging.error('Table creation failed, Reason: %s' % e)

# データ挿入
def insert_data():
    # データ挿入開始をログ出力
    logging.info('Start inserting data')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                data = [('John', 20), ('Lucy', 25), ('Tom', 30)]
                flattened_data = [d for item in data for d in item]
                cursor.executemany(INSERT_DATA_SQL, [flattened_data])
                conn.commit()
                # データ挿入成功をログ出力
                logging.info('Data insertion successful')
            except Exception as e:
                conn.rollback()
                # データ挿入失敗をログ出力、理由と共に
                logging.error('Data insertion failed, Reason: %s' % e)

# データクエリ
def select_data():
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            cursor.execute(SELECT_DATA_SQL)
            result = cursor.fetchall()
            for row in result:
                print(row)

# テーブル削除
def drop_table():
    # テーブル削除開始をログ出力
    logging.info('Start dropping the table')
    with pymysql.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cursor:
            try:
                cursor.execute(DROP_TABLE_SQL)
                conn.commit()
                # テーブル削除成功をログ出力
                logging.info('Table dropped successfully')
            except Exception as e:
                conn.rollback()
                # テーブル削除失敗をログ出力、理由と共に
                logging.error('Table drop failed, Reason: %s' % e)

if __name__ == '__main__':
    create_table()
    insert_data()
    select_data()
    drop_table()

# スクリプト実行完了をログ出力
logging.info('Script execution completed')

参照資料

• OceanBase Database への接続方法の詳細については、「概要」をご参照ください。