【Python】PostgreSQLに接続してSQL実行（SqlAlchemy）

qiitaに投稿したやつをこっちにも載せておきます。

やること

下記のようなテーブルからリストとして[[hoge1, hoge2], [hoge3, hoge4]]を取得します
＊その他SQL文であればなんでも実行できると思います

Column1	Column2
hoge1	hoge2
hoge3	hoge4

概要

• sqlalchemyでポスグレに接続してSQLを実行
• ソースはgithubに置きました（sqliteへの接続も有り）
• 使いまわしたいのでクラスである程度の事は吸収してもらう（クラスのお勉強）
• 今回はデータを一行ずつリスト形式で出力（[[hoge1, hoge2], [hoge3, hoge4]…]）

前提

• postgresのデータベース構築済み
• Pythonの基礎知識がある（クラス等）

ソース

DB情報は下記を例とします。

• ユーザー名: postgres
• password: postgres
• ホスト: localhost:5432
• データベース名: hoge
• テーブル名: hogehoge

from sqlalchemy import create_engine, text


class ConnectPostgreSQL:
    """
    PostgreSQLに接続するクラス

    Args:
        user_name (str): ユーザー名
        password (str): パスワード
        host (str): ホスト
        database_name (str): データベース名

    Methods:
        execute_sql: SQL文を直接実行
    """
    def __init__(
          self, user_name: str, password: str,
          host: str, database_name: str
    ):
        # DB接続に必要なエンジンを作成
        self.engine = create_engine(
            f"postgresql://{user_name}:{password}@{host}/{database_name}"
        )

    def execute_sql(self, sql: str, get_result: bool = False) -> list:
        """
        SQL文を直接実行

        Args :
            sql (str): SQL文
            return_result (bool): 結果を返すかどうかのフラグ。デフォルトはFalse
        """
        with self.engine.begin() as connection:
            input_sql = text(f"{sql}")
            result = connection.execute(input_sql)
            if get_result is True:
                sql_result = []
                for row in result:
                    sql_result.append(list(row))
                return sql_result


# ポスグレに接続
con_psg = ConnectPostgreSQL(
    "postgres", "postgres", "localhost:5432", "hoge"
)
sql = "SELECT * FROM hogehoge"
result = con_psg.execute_sql(sql, get_result=True)
# リストで結果を返す
print(result)

>>> [[hoge1, hoge2], [hoge3, hoge4]...]