sqlalchemyでpandasのdatetime型をMySQLに1行ずつInsertする方法

解決したい問題

sqlalchemyでMySQLにpandasのdatetime型を1行ずつInsertしたい場合があるかと思います。
そのときに、pandasのdatetime型をそのまま挿入しようとするとエラーになります。

例えば以下のようなサンプルコードを考えます。

from sqlalchemy import create_engine, Table, MetaData, Column, DateTime
import pandas as pd

# pandasのdatetime型の列を持つDataFrameの作成
df = pd.DataFrame(pd.to_datetime(["2023-01-01 00:00:00"]), columns=["sample_datetime"])

# sqlalchemyの設定
engine = create_engine('mysql+mysqlconnector://root:root@mysql_host/test_database')
metadata = MetaData()
table = Table('sample_table', metadata, Column('sample_datetime', DateTime))
metadata.create_all(engine)

# データを挿入
with engine.connect() as connection:
    for _, row in df.iterrows():
        connection.execute(table.insert().values(**row))

このとき、以下のようなエラーが表示されます。

sqlalchemy.exc.ProgrammingError: (mysql.connector.errors.ProgrammingError) Failed processing pyformat-parameters; Python 'timestamp' cannot be converted to a MySQL type
[SQL: INSERT INTO sample_table (sample_datetime) VALUES (%(sample_datetime)s)]
[parameters: {'sample_datetime': Timestamp('2023-01-01 00:00:00')}]

解決法

解決方法は至ってシンプルで、astype(str)によってdatetime型の列をstr型に変換すればうまくいきます。

df["sample_datetime"] = df["sample_datetime"].astype(str)

検証環境

• Python 3.9.12
• sqlalchemy 1.4.42
• MySQL 5.7

備考

なお

df.to_sql('sample_table', engine, if_exists='append', index=False)

のように記載して、一括してDataFrameをInsertするとエラーは発生しませんでした。
あくまで、DataFrameを1行ずつ処理したい場合の対処法ですので、ご留意ください。