分析ノート

とあるSDKを使って実装している処理で、利用しているAPIのエラーが頻発するようになり、
エラーが起きたらリトライする処理をスクラッチで作る必要があったのでその時の実装をメモしておきます。

今回は利用しているSDKのメソッドにリトライ処理が内包されておらず、
気軽にライブラリを追加できる環境でもなかったのでスクラッチで実装しましたが、
通常は他の手段がないか探すことをお勧めします。

例えば、requestsのようなライブラリは引数でリトライ回数を指定できますし、
世の中にはリトライを行う専用のライブラリなども転がっています。

この記事のコードはそれらが使えなかった場合の最後の手段です。

実際の処理はお見せできないので、サンプルとして一定確率(80%)エラーになる関数を作っておきます。

import numpy as np


def main_function():
    num = int(np.random.choice([0, 1], p=[0.8, 0.2]))
    return 1/num

さて、これを成功するまでリトライする関数を作ります。
実装するにあたって定めた要件は以下の通りです。

– 規定回数リトライする。(今回は5回と定めました。)
– 初回の実行と合わせて、実際に実行を試みるのは、{1+リトライ回数}回。
– 発生したエラーは都度表示する。
– 規定回数全てエラーになったら、最後に発生したエラーを呼び出し元に返す。
– エラーから次のリトライまでは、1秒,2秒,4秒,8秒,…と間隔をあける。

書いてみたのが次のコードです。

from time import sleep


max_retry_count = 5  # リトライ回数

for retry_count in range(max_retry_count+1):
    try:
        print(main_function())  # 実行したい処理
        break  # 成功したらループを抜ける
    except Exception as e:
        print(e)
        if retry_count == max_retry_count:
            print(f"規定回数({max_retry_count}回)のリトライに失敗しました。")
            raise
        # {2^リトライ回数}秒待ちを入れる
        dely = 2**retry_count
        print(f"{dely}秒後にリトライします。{retry_count+1}/{max_retry_count}回目。")
        sleep(dely)

２回のリトライ(３回目の試行)で成功すると出力は以下のようになります。

division by zero
1秒後にリトライします。1/5回目。
division by zero
2秒後にリトライします。2/5回目。
1.0

最後まで失敗すると以下のようになります。

division by zero
1秒後にリトライします。1/5回目。
division by zero
2秒後にリトライします。2/5回目。
division by zero
4秒後にリトライします。3/5回目。
division by zero
8秒後にリトライします。4/5回目。
division by zero
16秒後にリトライします。5/5回目。
division by zero
規定回数(5回)のリトライに失敗しました。
---------------------------------------------------------------------------
ZeroDivisionError                         Traceback (most recent call last)
 in 
      7 for retry_count in range(max_retry_count+1):
      8     try:
----> 9         print(main_function())  # 実行したい処理
     10         break  # 成功したらループを抜ける
     11     except Exception as e:

 in main_function()
      3 def main_function():
      4     num = int(np.random.choice([0, 1], p=[0.8, 0.2]))
----> 5     return 1/num

ZeroDivisionError: division by zero

ちゃんとエラーになりましたね。

PyMySQLの公式ドキュメントのExamplesで使われている、
cursorclass=pymysql.cursors.DictCursor
の話です。

import pymysql.cursors

# Connect to the database
connection = pymysql.connect(
    host='localhost',
    user='user',
    password='passwd',
    database='db',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor # これの話。
)

ドキュメントにも詳しい説明はないし、pep249でも言及されていないようなのでPyMySQLのソースコードも含めて調べてみました。

結論から言えば、実用上は黙って cursorclass=pymysql.cursors.DictCursor を指定しておけばよく、この記事は無駄知識の類のものになります。

まず、cursorclass に指定できる値は pymysql.cursors.DictCursor 以外に何があるのかですが、
これは、こちらのファイルで定義されている4種類のクラスが指定できます。
Github: pymysql/cursors.py

– pymysql.cursors.Cursor
– pymysql.cursors.DictCursor
– pymysql.cursors.SSCursor
– pymysql.cursors.SSDictCursor

https://github.com/PyMySQL/PyMySQL/blob/master/pymysql/connections.py#L179
に、

class Connection:
    # 中略
    def __init__(
        # 中略
        cursorclass=Cursor,

とある通り、デフォルトは、 pymysql.cursors.Cursor です。

この４種類のカーソルの違いですが、
Cursor と SSCursor は結果をタプルで返し、
DictCursor と SSDictCursor は結果を辞書で返してくれます。
結果の形に、CursorとSSCursor、DictCursorとSSDictCursorの間にはそれぞれ違いはありません。

SSとつく二つの方ですが、これらは主にデータが非常に大きいときや、ネットワークが遅いときなどに使います。
SSCursorのコメントがわかりやすいですね。

Unbuffered Cursor, mainly useful for queries that return a lot of data,
or for connections to remote servers over a slow network.
Instead of copying every row of data into a buffer, this will fetch
rows as needed. The upside of this is the client uses much less memory,
and rows are returned much faster when traveling over a slow network
or if the result set is very big.
There are limitations, though. The MySQL protocol doesn’t support
returning the total number of rows, so the only way to tell how many rows
there are is to iterate over every row returned. Also, it currently isn’t
possible to scroll backwards, as only the current row is held in memory.

個人的な感想としては最近の端末には十分なメモリが搭載されていて、数百万行単位のレコードを扱うときも、SS無しの方で十分さばけているので、
とりあえず DictCursor を使って、本当にメモリ不足で困った時だけ SSDictCursor を検討したらいいのかなと思っています。

Cursor(タプル) と DictCursor(辞書)についてはそれぞれ実行して結果を比較しておきましょう。

まず、cursorclassにpymysql.cursors.Cursorを指定した(もしくは何も指定しなかった場合)の結果です。
テーブルは以前の記事で作ったやつをそのまま使います。

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users"
        cursor.execute(sql)
        result = cursor.fetchall()
print(result)
"""
(
    (1, 'webmaster@python.org', 'very-secret'),
    (2, 'sato@python.org', 'very-secret'),
    (3, 'suzuki@python.org', 'very-secret'),
    (4, 'takahashi@python.org', 'very-secret'),
    (5, 'tanaka@python.org', 'very-secret')
)
"""

1レコードごとに結果がタプルで戻っているだけでなく、fetchallすると戻り値はタプルのタプルになっていますね。
戻ってきた結果の各値が、SELECT句のどの列の値なのかが明示されていないので、自分でマッピングする必要があります。
正直これは少し使いにくいです。

続いて、pymysql.cursors.DictCursor を指定した場合の結果です。

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users"
        cursor.execute(sql)
        result = cursor.fetchall()
print(result)
"""
[
    {'id': 1, 'email': 'webmaster@python.org', 'password': 'very-secret'},
    {'id': 2, 'email': 'sato@python.org', 'password': 'very-secret'},
    {'id': 3, 'email': 'suzuki@python.org', 'password': 'very-secret'},
    {'id': 4, 'email': 'takahashi@python.org', 'password': 'very-secret'},
    {'id': 5, 'email': 'tanaka@python.org', 'password': 'very-secret'}
]
"""

ご覧の通り、1レコードごとに「列名:値」の辞書として値が得られ、それらの配列として結果が返されます。
各値がSELECT句のどの列のものなのかはっきりしているのでとても便利です。
また、このままpandasのデータフレームに変換することもできます。
通常はこれを使えば良いでしょう。

※この記事ではDBはAWSのAuroraを想定しています。

Auroraに多くのデータをまとめて登録したいとき、1レコードごとにINSERT文を書いて実行するのは少しイケてない思っていたので、
良い方法を探していたところ、 executemany というメソッドが用意されているのを知りました。
pep249 にも記載があるので、PyMySQL以外のライブラリでも用意されていると思います。

サンプルコード載せるに状況の説明です。
INSERTしたいデータは例えば次のようなものだとします。

print(df)
"""
           date      open      high       low     close
0    2020-07-08  22481.61  22667.95  22438.65  22438.65
1    2020-07-09  22442.30  22679.08  22434.38  22529.29
2    2020-07-10  22534.97  22563.68  22285.07  22290.81
3    2020-07-13  22591.81  22784.74  22561.47  22784.74
4    2020-07-14  22631.87  22677.02  22538.78  22587.01
..          ...       ...       ...       ...       ...
195  2021-04-23  28939.12  29035.34  28770.62  29020.63
196  2021-04-26  29095.49  29241.28  28896.37  29126.23
197  2021-04-27  29174.53  29187.11  28990.19  28991.89
198  2021-04-28  28935.51  29139.70  28875.91  29053.97
199  2021-04-30  28996.66  29046.49  28760.27  28812.63

[200 rows x 5 columns]
"""

そして、INSERT先のテーブルは以下です。
(connection は既にRDSへの接続が完了しているものとします。)

with connection.cursor() as cursor:
    cursor.execute("SHOW CREATE TABLE nikkei")
    result = cursor.fetchone()
    
print(result["Create Table"])
"""
CREATE TABLE `nikkei` (
  `date` date NOT NULL,
  `open` double NOT NULL,
  `high` double NOT NULL,
  `low` double NOT NULL,
  `close` double NOT NULL,
  PRIMARY KEY (`date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
"""

通常であれば、この200あるレコード一つ一つについて、for文で回して、それぞれINSERT文を発行するところなのですが、
cursor.executemany を使うと次のように(表向きは)一発でINSERTできます。
(おそらく内部では1レコードずつINSERT文が発行されているのですが。)

insert_sql = """
    INSERT INTO
        nikkei (
            date,
            open,
            high,
            low,
            close
        )
    VALUES
        (
            %s, %s, %s, %s, %s
        )
"""

with connection.cursor() as cursor:
    cursor.executemany(insert_sql, df.values.tolist())
    connection.commit()

タイトルを逆にいうとPyMySQLのプレースホルダには qmark(?), numeric(:1), named(:name) は実装されてないようです。

※ この記事は、 version 1.0.2 の時点の PyMySQLについて書いています。将来的に他のプレースホルダーも実装されるかもしれません。

!pip freeze | grep PyMySQL
PyMySQL==1.0.2

pep249では、paramstyle として次の5つが定められています。

PyMySQLはpep249に従って実装さているので、この5つが全部使えるのかなと思って試したら、そうなってないことに気づきました。
タイトルにも書きました通り、実装されているのは format と pyformat だけです。
とはいえ、実用的なことを考えると、各値に名前をつけないときは、format、名前をつけたいときはpyformatを使えばいいので、この二つで十分だと思います。

実際にやってみます。
テーブルは前の記事で作ったものをそのまま使います。

まずDBに接続します。接続情報は各自の環境の値をご利用ください。

import pymysql.cursors

con_args = {
    "host": "{RDSのエンドポイント/同サーバーのDB場合はlocalhost}",
    "user": "{ユーザー名}",
    "password": "{パスワード}",
    "database": "{DB名}",
    "cursorclass": pymysql.cursors.DictCursor
}

connection = pymysql.connect(**con_args)

まず、 format スタイルから。サンプルコードなどで頻繁に見かけるのはこれです。

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users WHERE email=%s"
        cursor.execute(sql, ('webmaster@python.org',))
        result = cursor.fetchone()
        print(result)

# {'id': 1, 'email': 'webmaster@python.org', 'password': 'very-secret'}

続いて、　pyformat。　このサンプルコードではありがたみがないですが、多くのプレースホルダーを使うクエリでは可読性向上に期待できます。

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users WHERE email=%(email)s"
        cursor.execute(sql, {"email": "webmaster@python.org"})
        result = cursor.fetchone()
        print(result)

# {'id': 1, 'email': 'webmaster@python.org', 'password': 'very-secret'}

qmark だと cursor.execute でエラーになります。 (エラーになるので例外処理入れてます。)

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users WHERE email=?"
        try:
            cursor.execute(sql, ('webmaster@python.org',))
            result = cursor.fetchone()
            print(result)
        except Exception as e:
            print(e)
            # not all arguments converted during string formatting

numeric も同様。

with connection.cursor() as cursor:
        sql = "SELECT id, email, password FROM users WHERE email=:1"
        try:
            cursor.execute(sql, ('webmaster@python.org',))
            result = cursor.fetchone()
            print(result)
        except Exception as e:
            print(e)
            # not all arguments converted during string formatting

named もエラーになります。これだけエラー文が違います。

with connection.cursor() as cursor:
        try:
            sql = "SELECT id, email, password FROM users WHERE email=:email"
            cursor.execute(sql, {"email": "webmaster@python.org"})
            result = cursor.fetchone()
            print(result)
        except Exception as e:
            print(e)
            # (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near ':email' at line 1")

今回の記事では、PythonのコードでSQLを実行しAuroraを操作する方法を紹介します。
AuroraはMySQL互換のインスタンスを使っているので、PythonのライブラリはMySQL用のものを利用します。
複数種類あるのですが、僕はPyMySQLを利用することが多いです。
(職場で先輩が使ってたというだけの理由で使い続けており、他のライブラリと比較検証したわけじゃないので、
いつかちゃんとメリットデメリット比較したいです。)

PyMySQL のドキュメントはこちら
PyMySQL · PyPI
前回の記事でも書きましたが、サンプルコードが軽く記載してあるだけで、あまり詳しい説明などはありません。
pep249を必要に応じて参照する必要があります。

といっても、使い方は簡単なので、ドキュメントのサンプルコードだけあれば実用上はなんとか使っていけます。
(DBヘの接続、SQLの実行と結果の取り出しができれば十分。)

この記事では、サンプルコードを参考に、最低限の操作をやってみようと思います。

まず、DBにサンプルのTableを作っておきます。

> CREATE TABLE users (
    id int(11) NOT NULL AUTO_INCREMENT,
    email varchar(255) NOT NULL,
    password varchar(255) NOT NULL,
    PRIMARY KEY (`id`)
);

Query OK, 0 rows affected (0.05 sec)

まず、DBへの接続です。
DBのエンドポイント/ユーザー名/パスワード/DB名 をそれぞれ設定し、接続をします。
cursorclass=pymysql.cursors.DictCursor を設定するのは、結果をDict形で受け取るためです。
(これは必須ではないのですが、Dictで受け取った方が後々扱いやすいです。)
{}で書いてる部分は各自の環境の値を入れてください。

import pymysql.cursors


con_args = {
    "host": "{RDSのエンドポイント/同サーバーのDB場合はlocalhost}",
    "user": "{ユーザー名}",
    "password": "{パスワード}",
    "database": "{DB名}",
    "cursorclass": pymysql.cursors.DictCursor
}

connection = pymysql.connect(**con_args)

新規レコードを挿入してみます。
SQLインジェクション対策にプレースホルダ(%s)を使います。
自動コミットではないので、 connection.commit() を忘れないようにするのがポイントです。

with connection.cursor() as cursor:
    sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
    cursor.execute(sql, ('webmaster@python.org', 'very-secret'))
    connection.commit()

次はSELECT文です。 1レコード取り出すサンプルコードは次のようになります。

with connection.cursor() as cursor:
    sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
    cursor.execute(sql, ('webmaster@python.org',))
    result = cursor.fetchone()
    print(result)

# 結果
# {'id': 1, 'password': 'very-secret'}

ドキュメントのサンプルコードはこれだけですね。

ほとんど代わり映えしないのですが、結果が複数行になるSELECT文も試しておきましょう。
まずレコード自体が1行しかないとどうしようもないのでデータを増やします。

with connection.cursor() as cursor:
    sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
    cursor.execute(sql, ('sato@python.org', 'very-secret'))
    cursor.execute(sql, ('suzuki@python.org', 'very-secret'))
    cursor.execute(sql, ('takahashi@python.org', 'very-secret'))
    cursor.execute(sql, ('tanaka@python.org', 'very-secret'))
    connection.commit()

結果が複数行になる場合は、fetchone()の代わりに、fetchall()を使います。

with connection.cursor() as cursor:
        sql = "SELECT id, email FROM users"
        cursor.execute(sql)
        result = cursor.fetchall()
        print(result)

# [{'id': 1, 'email': 'webmaster@python.org'}, {'id': 2, 'email': 'sato@python.org'}, 
# {'id': 3, 'email': 'suzuki@python.org'}, {'id': 4, 'email': 'takahashi@python.org'},
# {'id': 5, 'email': 'tanaka@python.org'}]

ちなみにこの結果の型ですが、pandasのDataFrameに簡単に変換できるので便利です。

import pandas as pd


df = pd.DataFrame(result)
print(df)
"""
   id                 email
0   1  webmaster@python.org
1   2       sato@python.org
2   3     suzuki@python.org
3   4  takahashi@python.org
4   5     tanaka@python.org
"""

もちろんfetchoneを繰り返して1行ずつ取り出すこともできるのですが、個人的にはfetchallでさっさと取り出して
データフレームにしてしまった方が扱いやすいと思います。
fetchoneで順番に取り出すとしたら次のようなコードでしょうか。

with connection.cursor() as cursor:
        sql = "SELECT id, email FROM users"
        cursor.execute(sql)

row = cursor.fetchone()
while row is not None:
    print(row)
    row = cursor.fetchone()
"""
{'id': 1, 'email': 'webmaster@python.org'}
{'id': 2, 'email': 'sato@python.org'}
{'id': 3, 'email': 'suzuki@python.org'}
{'id': 4, 'email': 'takahashi@python.org'}
{'id': 5, 'email': 'tanaka@python.org'}
"""

次のような書き方でも動きます。

with connection.cursor() as cursor:
    sql = "SELECT id, email FROM users"
    cursor.execute(sql)

for row in cursor:
    print(row)
    
"""
{'id': 1, 'email': 'webmaster@python.org'}
{'id': 2, 'email': 'sato@python.org'}
{'id': 3, 'email': 'suzuki@python.org'}
{'id': 4, 'email': 'takahashi@python.org'}
{'id': 5, 'email': 'tanaka@python.org'}
"""

PEP 249 — Python Database API Specification v2.0と言うものがあることを最近知ったのでその紹介です。

お恥ずかしい話ですが正直に言うと、これまでPython経由でデータベースを操作するコードを書くときは、
誰かが既に書いたコードや本のサンプルなどを参照して、DBヘの接続やSQLの実行部分はほとんどコピーして使っていました。

頻繁に実施する処理ではあるので、このブログにもまとめておこうと思い、
ブログ記事にするからには仕様を正しく理解してから書きたいと考えて調べてたら行きあたったのがこのPEP249です。

これは何かというと、リレーショナルデータベースにアクセスする機能を提供する
Pythonモジュールが似たような作りになるようにAPI仕様を定義したものです。

PEP249のページでも
This API has been defined to encourage similarity between the Python modules that are used to access databases.
と書いてありますね。

なぜこんなものが必要になるかというと、RDBには多くの種類があり、そしてそれぞれのRDBごとにそれを利用するモジュールが作られたとして、
それらが全部バラバラに仕様を決めて実装されていたら、使いにくいし、もし別のDBやモジュールに移行したくなったときに使い勝手が違うと大変だからです。

そこで、PEP249 (DB-API 2.0)として、DB接続や、SQLの実行、結果の取り出しなどのAPIの仕様を定めてくれています。
実際、多くのモジュールがこれに沿って実装されているようです。

PEP249 のページを読んでいくと、DBの接続に使う connect メソッドや、　Connection Object、
SQLを実行するexecuteメソッドや、結果を取り出す、fetchone/ fetchall など見慣れた(コピペで使ってた)メソッド群の説明が書いてあります。

PyMySQL などのドキュメントを見てると、
サンプルコードが少し書いてあるだけで、各メソッドの説明がほとんどなく非常に不親切だなと前々から思っていたのですが、
PEP249に沿って実装されているのであれば話は別です。
これらのモジュールはPEP249を読んで使うものだったのですね。