PyMySQLでSQLインジェクション対策を実施する方法

概要

対象SQLのようなクエリをPyMySQLを使いSQLインジェクション対策実施した上で実行する方法について書いています

方法

対象SQL

単純なクエリの場合

SQL

SELECT content
FROM article_cache
WHERE
crawl_jobid = '1111'

Python

import pymysql
from pymysql.cursors import DictCursor

# MySQLに接続
connection = pymysql.connect(
    host='your_host',
    user='your_username',
    password='your_password',
    db='your_database',
    charset='utf8',
    cursorclass=DictCursor
)

# 対象の値
crawl_jobid = '1111'

try:
    with connection.cursor() as cursor:
        sql = f"""
        SELECT content
        FROM article_cache
        WHERE
        crawl_jobid = %s
        """

        # クエリの作成
        query = cursor.mogrify(sql, (crawl_jobid)) 

        # クエリ実行
        cursor.execute(query)

        # 結果の取得
        result = cursor.fetchall()

finally:
    connection.close()

IN句の場合

SQL

SELECT content
FROM article_cache
WHERE
crawl_jobid IN ('1111','2222','3333','4444')

Python

import pymysql
from pymysql.cursors import DictCursor

# MySQLに接続
connection = pymysql.connect(
    host='your_host',
    user='your_username',
    password='your_password',
    db='your_database',
    charset='utf8',
    cursorclass=DictCursor
)

# 対象の値
crawl_jobids = ['1111', '2222', '3333', '4444']

# 対象の値に数だけプレースホルダを作成
format_strings = ','.join(['%s'] * len(crawl_jobids))

try:
    with connection.cursor() as cursor:
        sql = f"""
        SELECT content
        FROM article_cache
        WHERE
        crawl_jobid IN ({format_strings})
        """

        # クエリの作成
        # リストの各要素を個別の引数として渡すため変数の先頭に「*」をつけています
        query = cursor.mogrify(sql, (*crawl_jobids)) 

        # クエリ実行
        cursor.execute(query)

        # 結果の取得
        result = cursor.fetchall()

finally:
    connection.close()

ORを使い複数指定する場合

SQL

SELECT content
FROM article_cache
WHERE
message LIKE '%' || 'aaaa' || '%' OR message LIKE '%' || 'bbbb' || '%' OR message LIKE '%' || 'cccc' || '%')

Python

import pymysql
from pymysql.cursors import DictCursor

# MySQLに接続
connection = pymysql.connect(
    host='your_host',
    user='your_username',
    password='your_password',
    db='your_database',
    charset='utf8',
    cursorclass=DictCursor
)

# 対象の値
search_terms = ['aaaa', 'bbbb', 'cccc']

# 対象の値に数だけプレースホルダを作成
like_strings = " OR ".join(["message LIKE %s" for _ in search_terms])
search_terms = ["%" + term + "%" for term in search_terms]

try:
    with connection.cursor() as cursor:
        sql = f"""
        SELECT content
        FROM article_cache
        WHERE
        {like_strings}
        """

        # クエリの作成
        # リストの各要素を個別の引数として渡すため変数の先頭に「*」をつけています
        query = cursor.mogrify(sql, (*search_terms)) 

        # クエリ実行
        cursor.execute(query)

        # 結果の取得
        result = cursor.fetchall()

finally:
    connection.close()