環境

python3.8.10
PostgreSQL 12.14　(少し古い)
pandas 1.3.5
SQLAlchemy 1.4.46

あらすじ

pandasのDataFrameをそのままテーブルとしてデータベースに格納したい。
そんな時、1つの方法としてSQLAlchemyを使う方法がある。
しかし、テーブルにする時に明示的に型指定をしていなかったために他のクエリでエラーを吐いてしまった。

DataFrame → テーブル

import pandas as pd
import psycopg2
from psycopg2.extras import DictCursor
from psycopg2.extras import execute_values
from sqlalchemy import create_engine


def insert_table_from_dataframe(df, table_name):
    """
    dataframeをDBに格納
    """
    # データベースの接続情報
    connection_config = {
        'user': 'ユーザー名',
        'password': 'パスワード', 
        'host': 'ホスト名',
        'port': 'ホストが指定するポート番号', 
        'database': 'データベース名'
    }

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

    # PostgreSQLに書き込む
    df.to_sql(table_name, con=engine, if_exists='append', index=False)
    return

こんな形でデータベースに接続してテーブルに書き込む。
しかし、これだと型をどのコードでも指定しておらず、カラムの型が何になっているか不明。
実際に確認すると、大体はよしなにやってくれるのだが、IPアドレスをINET型で格納して欲しいのにTEXT型になってしまった。

解決法

from sqlalchemy.dialects.postgresql import INET

# PostgreSQLに書き込む
df.to_sql(table_name, con=engine, if_exists='append', index=False, dtype={'ip': INET})

まず、sqlalchemyではデータベースに合わせて型を読み込むことができる。
全てのデータベースで共通して利用できる型名があり、それは https://docs.sqlalchemy.org/en/14/core/type_basics.html に書いてあるので参照されたい。
その場合には、from sqlalchemy import typesを加え、types.Integer()のような形で指定する。

データベース固有の表現を使いたい場合は、上記のようにデータベースごとにある型を読み込み、dtypeでカラム名とセットで指定する。
dialectsについては https://docs.sqlalchemy.org/en/20/dialects/ に詳細がある。