PyMySQLでBulk Updateをする方法と注意点【Python】

結論

• INSERT ... ON DUPLICATE KEY UPDATE 構文の場合はexecutemanyを使う。
• ELT&FIELDでのUpdateをする場合はプリペアドステートメントでパラメータをセットする。

INSERT ... ON DUPLICATE KEY UPDATE 構文

VALUESなどの一度にデータを挿入するクエリではexecutemanyメソッドを使うことができます。

conn = pymysql.connect(
    mysql_endpoint,
    user=username,
    passwd=password,
    db=dbname
)

def bulk_insert_and_update_users():
    prepared_statement = [
        [
            1, # id
            'Qiita太郎', # name
            20 # age
        ],
        [
            2,
            'Qiita花子',
            18
        ]
    ]

    with conn.cursor() as cursor:
        sql = (
            'INSERT INTO users '
            '(id, name, age) '
            'VALUES (%s, %s, %s) '
            'ON DUPLICATE KEY UPDATE '
            'name = VALUES(name), '
            'age = VALUES(age)'
        )
        cursor.executemany(sql, prepared_statement)

    conn.commit()

これを実行するとこのようなクエリが生成されます。

INSERT INTO users (id, name, age)
VALUES 
  (`1`, `Qiita太郎`, `20`),
  (`2`, `Qiita花子`, `18`)
ON DUPLICATE KEY UPDATE
  name = VALUES(name),
  age = VALUES(age);

シンプルで分かりやすいですね。

ELT & FIELD

INSERTをしない場合やINSERT ... ON DUPLICATE KEY UPDATEによるauto_incrementの問題が気になる場合はこちらを使用することになると思います。(参考)
その場合、いくつか注意しなくてはいけない点があるので見ていきましょう。

conn = pymysql.connect(
    mysql_endpoint,
    user=username,
    passwd=password,
    db=dbname
)

def bulk_update_users():
    records = [
        {
            'user_id': 1,
            'user_name': 'Qiita太郎',
            'user_age': 20
        },
        {
            'user_id': 2,
            'user_name': 'Qiita花子',
            'user_age': 18
        }
    ]

    id_list = []
    name_list = []
    age_list = []
​
    for record in records:
        id_list.append(str(record['user_id']))
        name_list.append(record['user_name'])
        age_list.append(record['user_age'])
​
    id_strings = ','.join(['%s'] * len(id_list))
    name_strings = ','.join(['%s'] * len(name_list))
    age_strings = ','.join(['%s'] * len(age_list))
​
    sql = (
        'UPDATE users SET '
        'name = '
        'ELT(FIELD(id, %(user_ids)s), %(user_names)s), '
        'age = '
        'ELT(FIELD(id, %(user_ids)s), %(user_ages)s) '
        'WHERE id IN (%(user_ids)s);'
        '' % dict(user_ids=id_strings,
                  user_names=name_strings,
                  user_ages=age_strings)
    )
​
    prepared_statement = tuple(id_list) \
        + tuple(name_list) \
        + tuple(id_list) \
        + tuple(age_list) \
        + tuple(id_list)
​
    with conn.cursor() as cursor:
        cursor.execute(sql, prepared_statement)

    conn.commit()

特徴的なのはこの部分ですね。

id_list = []
name_list = []
age_list = []
​
for record in records:
    id_list.append(str(record['user_id']))
    name_list.append(record['user_name'])
    age_list.append(record['user_age'])
​
id_strings = ','.join(['%s'] * len(id_list))
name_strings = ','.join(['%s'] * len(name_list))
age_strings = ','.join(['%s'] * len(age_list))
​
sql = (
    'UPDATE users SET '
    'name = '
    'ELT(FIELD(id, %(user_ids)s), %(user_names)s), '
    'age = '
    'ELT(FIELD(id, %(user_ids)s), %(user_ages)s) '
    'WHERE id IN (%(user_ids)s);'
    '' % dict(user_ids=id_strings,
              user_names=name_strings,
              user_ages=age_strings)
)

これを実行すると変数sqlには以下のような文字列が格納されます。

UPDATE users SET 
  name = ELT(FIELD(id, %s, %s), %s, %s), 
  age = ELT(FIELD(id, %s, %s), %s, %s)
WHERE id IN (%s, %s);

今回はわかりやすくするために助長に書いていますが、実際には%sの数は全て同じなのでrecordsの要素数から文字列を生成して使いまわしても問題ないです。

そして%演算子で挿入する順番で全てのパラメータを結合しています。

prepared_statement = tuple(id_list) \
    + tuple(name_list) \
    + tuple(id_list) \
    + tuple(age_list) \
    + tuple(id_list)

# (1, 2, `Qiita太郎`, `Qiita花子`, 1, 2, 20, 18, 1, 2)

これをconn.executeの引数に渡すことで以下のようなクエリが実行されます。

UPDATE users SET 
  name = ELT(FIELD(id, `1`, `2`), `Qiita太郎`, `Qiita花子`), 
  age = ELT(FIELD(id, `1`, `2`), `20`, `18`)
WHERE id IN (`1`, `2`);

なんでこんな面倒なことするの？

ただ単にクエリを作るだけなら文字列結合してしまえば簡単です。
しかしこれだとSQLインジェクションに対する脆弱性があります。

例えばユーザーの名前を"'; DROP TABLE users; '"に設定されてしまった場合に防御のしようがありません。

ダメな例

id_list = []
name_list = []
age_list = []
​
for record in records:
    id_list += [f"`{str(record['user_id'])}`"]
    name_list += [f"`{record['user_name']}`"]
    age_list += [f"`{record['user_age']}`"]

id_list_s = ",".join(id_list)
name_list_s = ",".join(name_list)
age_list_s = ",".join(age_list)

sql = (
    'UPDATE users SET '
    'name = '
    f'ELT(FIELD(id, {id_list_s}), {name_list_s}), '
    'age = '
    f'ELT(FIELD(id,{id_list_s}), {age_list_s}) '
    f'WHERE id IN ({id_list_s});'
)

一応replaceなどで'を\'に置き換えるという対策もできなくはないですが、execute実行時にpymysql側でエスケープしてくれるので素直にプリペアドステートメントでパラメータを挿入する方が良いでしょう。(参考)

おまけ

pymysqlで実際に実行されたSQL文を確認するにはexecute後にself.executedに保存されるので、このプロパティにアクセスすることで確認が可能です。

with conn.cursor() as cursor:
    sql = "SELECT * FROM users WHERE id = %s"
    cursor.execute(sql, (1))
    print(cursor._executed)

# SELECT * FROM users WHERE id = 1

参考資料

PyMySQL
10.5.5 MySQLCursor.executemany() Method
MySQL で ON DUPLICATE KEY UPDATE を使ってたら、カンスト(カウンターストップ)した話
imploding a list for use in a python MySQLDB IN clause
Pythonプリペアドステートメント。 SELECT INの問題