SQLAlchemy でクエリキャッシュを使う方法 (dogpile.cache編)

• 概要
• 環境
• サンプルコード
• ちょっと解説
• 最後に
• 参考サイト

概要

SQLAlchemy は以前 Beker cache という仕組みを使ってクエリキャッシュしていました
しかし SQLAlchemy 2.0 の現在ではその方法は廃止され dogpile.cache という汎用キャッシュの仕組みを使ってクエリキャッシュするのが王道になっています

前回 dogpile.cache の簡単な使い方を紹介しました
今回は SQLAlchemy と組み合わせて使ってみます

環境

• macOS 13.5.2
• Python 3.11.5
  • dogpile.cache 1.2.2
  • SQLAlchemy 2.0.21

サンプルコード

from dogpile.cache import make_region
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "user"

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


# Dogpileキャッシュリージョンを作成
cache_region = make_region().configure(
    "dogpile.cache.memory",  # メモリキャッシュを使用しますが、他のバックエンドも利用可能です
    expiration_time=3600,  # キャッシュの有効期限を設定（秒単位）
)


# SQLAlchemyセッションをキャッシュ対象にする
@cache_region.cache_on_arguments()
def get_user_by_id(user_id):
    db: Session = SessionLocal()
    print("Querying database for user:", user_id)
    result = db.query(User).filter_by(id=user_id).first()
    db.close()
    return result


# ユーザーを取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1:", cached_user.name)

# 同じユーザーを再度取得し、キャッシュを利用
cached_user = get_user_by_id(1)
print("User 1 (cached):", cached_user.name)

ちょっと解説

dogpile.cache は同一パラメータを使った関数呼び出しの場合はキャッシュを参照するようになります
なのでその仕組を使ってクエリを投げる関数を作成しその関数に対して cache_on_arguments アノテーションを付与することでキャッシュ化することができます

2回関数をコールしていますが2回目はキャッシュを参照しているため「Querying database for user: 1」が表示されないのが確認できます

こんな感じで既存のクエリを発行する関数を簡単にキャッシュ化することができます

最後に

dogpile.cache を使って SQLAlchemy のクエリキャッシュをする方法を紹介しました
アノテーション付与だけで使えるのでかなり簡単です

dogpile.cache には他にもキャッシュ戦略やキャッシュを特定するキーの生成方法などいろいろとカスタマイズすることができるので興味があれば参考サイトにある公式ドキュメントを見ることをおすすめします

参考サイト

• https://dogpilecache.sqlalchemy.org/en/latest/usage.html