SQLAlchemy1.4で非同期接続用のsessionを作る


ここではPostgreSQLに接続する例を示す。

必要なモジュールのインポート

from sqlalchemy.orm import sessionmaker # 同期接続の時の同じようにsessionmakerをインポート
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

create_engineの代わりにcreate_async_engineをインポートします。
非同期接続用のクラス、AsyncSessionをインポートします。

sessionを作成

// asyncpgなりpsycopg2なり
async_db = f"postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}:{PORT}/{DB_NAME}"

async_engine = create_async_engine(async_db)
async_session = sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
)

sessionmakerのコンストラクタを見ると、デフォルトではclass_にorm.session.Sessionが指定されているので、AsyncSessionを使うようにします。

https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/orm/session.py

あとは

async with async_session() as session:

contextmanagerからsessionをもらって使うだけです。

参考までに

asynccontextmanagerでwrapして例外時の後処理を記述しても良さそうです。

https://docs.python.org/ja/3/library/contextlib.html#contextlib.asynccontextmanager