pandas.read_sql()の中にSQL文を書くと"explicit type casts"というエラーが出る

注(2021/09/20 25:40)

コメントにてSQLインジェクション対策についてご指摘をいただき、
それに従って内容を追記しました（本文下部の追記をご確認願います。）

PostgreSQLからデータを取得する際、
pandas.read_sql()の中にSQL文を書くと"explicit type casts"というエラーが出る現象に悩まされていました。

エラー発生.py

import pandas as pd
from sqlalchemy import create_engine

connection_config ={
    'user':'workuser',
    'password':'********',
    'host':'localhost',
    'port':'5432',
    'database':'database_test'
}


engine = create_engine(f'postgresql://{user}:{password}@{host}:{port}/{database}')

df_a = pd.read_csv("sample.csv")
sample_id = df_a["id"].iloc[-1]
# 00000000と仮定

df_b = pd.read_sql(f'SELECT * FROM sample_table WHERE id = {sample_id}', con=engine)

LINE 1: SELECT * FROM sample_table WHERE id = 00000000

HINT: No operator matches the given name and argument type(s). You might need to add explicit type casts. [SQL: 'SELECT * FROM sample_table WHERE id = 00000000']

このエラーに対して、次のstackoverflowの記事を元に修正したところ、
解決しました。

[Python Dataset module error: You might need to add explicit type casts]
(https://stackoverflow.com/questions/31829269/python-dataset-module-error-you-might-need-to-add-explicit-type-casts/)

次のようにSQL文を二重引用符で囲い、{sample_id}を一重引用符で囲うとエラーが消えました。

エラー解消（該当部分のみ）.py

df_b = pd.read_sql(f"SELECT * FROM sample_table WHERE id = '{sample_id}'", con=engine)

追記

次のURLによればpd.read_sql()の（）内にSQL文を書くと
SQLインジェクション対策ができないとのことです。

Python pandas.read_sql SQL injection safe?

こちらの内容に従い、内容を追記いたします。

エラー解消（SQLインジェクション対策追加）.py

import pandas as pd
import psycopg2

connection_config ={
    'user':'workuser',
    'password':'********',
    'host':'localhost',
    'port':'5432',
    'database':'database_test'
}

url = 'postgresql://{user}:{password}@{host}:{port}/{database}'.format(**connection_config)

df_a = pd.read_csv("sample.csv")
sample_id = df_a["id"].iloc[-1]
# 00000000と仮定

try:
    conn = psycopg2.connect(url)
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM experiment WHERE id = %s;" , (sample_id,))
        result = cursor.fetchall()
        df_b = pd.DataFrame(result)

except Exception as e:
    print(e)
finally:
    conn.close()