高スループットを要求されるWebサービスでは、データベースI/O処理の非同期化がパフォーマンス向上の鍵となります。aiomysqlはasyncioフレームワークと連携し、MySQLサーバーとの非ブロッキング通信を実現するPythonライブラリです。本稿では接続管理からトランザクション処理まで、実践的な実装パターンを解説します。

非同期データベース処理の必要性

従来の同期型データベースドライバでは、I/O待ちでスレッドがブロックされスケーラビリティに限界が生じます。aiomysqlはコルーチンベースのAPIを提供し、ネットワーク待ち時間中に他のタスクを処理可能にします。特にaiohttpやFastAPIといった非同期Webフレームワークとの組み合わせで真価を発揮します。

環境構築のポイント

Python 3.7以降を前提にインストールを行います。MySQLサーバーのバージョンに応じて適切な文字セットを指定することが重要です。

pip install aiomysql==0.0.22
# MySQL 8.0以降を利用する場合
pip install 'aiomysql[cryptography]'

接続プールの最適化設定

接続リソースの効率的な管理にはプールサイズの調整が不可欠です。以下の実装例では最小/最大接続数を明示的に指定しています。

import aiomysql

async def initialize_connection_pool():
    return await aiomysql.create_pool(
        host='localhost',
        port=3306,
        user='service_account',
        password='encrypted_pass',
        db='application_db',
        charset='utf8mb4',
        minsize=3,
        maxsize=15,
        autocommit=False
    )

安全なデータ取得パターン

パラメータ化クエリによるSQLインジェクション対策と、結果セットの適切な処理方法を示します。

async def retrieve_active_users(db_pool, status=1):
    async with db_pool.acquire() as connection:
        async with connection.cursor(aiomysql.DictCursor) as cursor:
            await cursor.execute(
                "SELECT user_id, email FROM accounts WHERE is_active = %s",
                (status,)
            )
            return await cursor.fetchall()

トランザクションの堅牢な実装

例外処理とロールバックを組み合わせた、信頼性の高いトランザクション処理の実装例です。

async def process_user_registration(db_pool, user_data):
    async with db_pool.acquire() as conn:
        async with conn.cursor() as cur:
            try:
                await conn.begin()
                await cur.execute(
                    "INSERT INTO profiles (name, email) VALUES (%s, %s)",
                    (user_data['name'], user_data['email'])
                )
                await cur.execute(
                    "INSERT INTO credentials (user_id, hash) VALUES (%s, %s)",
                    (cur.lastrowid, user_data['password_hash'])
                )
                await conn.commit()
            except Exception as e:
                await conn.rollback()
                raise RuntimeError(f"登録処理に失敗: {str(e)}") from e

Webフレームワークとの統合例

aiohttpアプリケーションに組み込む際の初期化パターンを紹介します。

from aiohttp import web

async def setup_application():
    app = web.Application()
    app.cleanup_ctx.append(init_db_context)
    app.router.add_get('/users', handle_user_request)
    return app

async def init_db_context(app):
    app['db_pool'] = await initialize_connection_pool()
    yield
    app['db_pool'].close()
    await app['db_pool'].wait_closed()