Python で 'echo' メッセージをカスタマイズ：SQLAlchemy エンジン

2024-07-27

SQLAlchemy エンジン 'echo' メッセージのフォーマット変更方法

SQLAlchemy エンジンの echo 属性は、実行される SQL クエリに関する情報を標準出力に出力します。デフォルトのフォーマットは簡潔ですが、必要に応じて詳細な情報を含めるように変更できます。

方法

echo 属性に logging.Formatter オブジェクトを設定します。
フォーマッターオブジェクトを使用して、出力メッセージのフォーマットを定義します。

例

from sqlalchemy import create_engine
from logging import Formatter

# エンジン作成
engine = create_engine("sqlite:///example.db", echo=True)

# フォーマッターオブジェクト作成
formatter = Formatter(
    fmt="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# エンジンにフォーマッターを設定
engine.echo.set_formatter(formatter)

# クエリ実行
engine.execute("SELECT * FROM users")

出力例

2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users

フォーマットオプション

logging.Formatter オブジェクトは、以下の属性を使用して出力メッセージのフォーマットをカスタマイズできます。

style: メッセージスタイル (デフォルトは '%(asctime)s %(levelname)-8s %(message)s')
datefmt: 日付フォーマット文字列
fmt: メッセージフォーマット文字列

SQLAlchemy は、echo 属性以外にも、ロギング機能を提供しています。詳細は、SQLAlchemy ドキュメントを参照してください。
より複雑なフォーマットや、メッセージをファイルに記録する場合は、独自のロギングハンドラーを実装する必要があります。
上記の例は、標準出力にメッセージを出力する単純な例です。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)

# エンジン作成
engine = create_engine("sqlite:///example.db", echo=True)

# フォーマッターオブジェクト作成
formatter = Formatter(
    fmt="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# エンジンにフォーマッターを設定
engine.echo.set_formatter(formatter)

# セッション作成
Session = sessionmaker(bind=engine)
session = Session()

# クエリ実行
users = session.query(User).all()

# データ追加
user = User(name="John Doe")
session.add(user)
session.commit()

# クエリ実行
users = session.query(User).all()

# セッションクローズ
session.close()

2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine BEGIN (implicit)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine COMMIT
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine BEGIN (implicit)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine INSERT INTO users (name) VALUES (?)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine COMMIT
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users

説明

sqlite:///example.db という名前の SQLite データベースに接続する SQLAlchemy エンジンを作成します。
セッションを作成し、User テーブルからすべてのレコードをクエリします。
新しい User オブジェクトを作成し、データベースに追加します。
セッションを閉じます。

実行

このコードを実行するには、Python 3 と SQLAlchemy がインストールされている必要があります。

python sample.py

出力

コードを実行すると、標準出力に以下のメッセージが表示されます。

2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine BEGIN (implicit)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine COMMIT
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine BEGIN (implicit)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine INSERT INTO users (name) VALUES (?)
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine COMMIT
2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users

from sqlalchemy import create_engine
from logging import DEBUG

# エンジン作成
engine = create_engine("sqlite:///example.db", echo=False)

# エンジンレベルのロギング設定
engine.logger.setLevel(DEBUG)

# クエリ実行
engine.execute("SELECT * FROM users")

2023-11-14 10:29:59 DEBUG sqlalchemy.engine.base.Engine SELECT * FROM users

独自のロギングハンドラーを実装する

logging.Handler サブクラスを継承した独自のロギングハンドラーを実装できます。

from sqlalchemy import create_engine
from logging import Handler, Formatter

class MyHandler(Handler):
    def __init__(self):
        super().__init__()

        # フォーマッターオブジェクト作成
        self.formatter = Formatter(
            fmt="%(asctime)s %(levelname)-8s %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )

    def emit(self, record):
        # 独自の処理
        # ...

        # フォーマットされたメッセージを出力
        msg = self.formatter.format(record)
        print(msg)

# エンジン作成
engine = create_engine("sqlite:///example.db", echo=False)

# 独自のロギングハンドラーを設定
handler = MyHandler()
engine.logger.addHandler(handler)

# クエリ実行
engine.execute("SELECT * FROM users")

サードパーティライブラリを使用する

sqlalchemy-echo などのサードパーティライブラリを使用して、echo メッセージのフォーマットをカスタマイズできます。

from sqlalchemy import create_engine
from sqlalchemy_echo import Echo

# エンジン作成
engine = create_engine("sqlite:///example.db", echo=False)

# Echo オブジェクト作成
echo = Echo(engine)

# フォーマット設定
echo.set_formatter(
    fmt="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# クエリ実行
echo.execute("SELECT * FROM users")