SQLAlchemyのexecute()を用いてSQL文を実行したい

はじめに

• sqlalchemyの難しそうなメソッドを利用せず、簡単にSQL文を実行したかった
• また、sqlalchemyを使っていて、何箇所か詰まるところがあったためご共有

Code sample

1. （任意）DBとコミュニケーションするclassを作る

• 使いまわしたり、何個もオブジェクトを作成する時はclass化しておくと便利です。（ログとかも）

from sqlalchemy import Connection, MetaData, create_engine, text

class DBClass:
    def __init__(self, schema)
    self.schema = schema
    self.engine = create_engine(
    "postgresql://{user}:{password}@{host}:{port}/{database}"
    )

    self.metadata = MetaData(schema=self.schema)
    self.metadata.reflect(bind=self.engine, schema=self.schema)

2. SQL文を実行する

• sqlalchemy.textを利用して、sql文をconnectionに対してexecute()する

sql = f"""
DELETE FROM {schema}.{table_name} WHERE {condition};
"""
dbc = DBClass(schema)
with dbc.engine.begin() as connection:
    results = connection.execute(text(sql))
    logger.info(f"delete {results.rowcount} record.")

sqlalchemyを使っていて詰まったポイント

• postgresから作ったはずのtable一覧が取得できない
• engine.connect()に対してexecute()を実行しても、反映されない
  • 具体的にはSELECT文は通るのにDELETE文は通らなかった
  • engine.coccect()　-> engine.begin()にすることで治った（理由不明）
    • 公式ドキュメントではengine.coccect()　を利用。
    • 理由がわかったら追記します。

sqlalchemyのオブジェクトがうまく動作しない時、試してみるべきポイント

• reflect（）が実行されているか確認する
  • (memo)engineオブジェクトを利用して何か処理を実行した後、reflect()を行わないと反映されない？(branch自体は存在するけれどgit fetchをしないと情報が取得できない、といった感じのものだろうか。)
• engine.connect()でなくengine.begin()を利用してみる

おわりに

• SQLAlchemyの公式docがあるが、読み解くのは難しい...。