Python の同期コードを async/await へ安全リファクタするプロンプト
いつ使うか
requests / time.sleep / 同期 ORM で書かれた Python スクリプトを、httpx / asyncio / async ORM に置き換えたい時。雑に async def を付けるだけだと、内部で同期 I/O が走ってイベントループをブロックする「偽 async」になりがち。構造的に安全な置換手順を踏ませる。
プロンプト本文 (コピペして使う)
あなたは Python の async プログラミング専門家です。以下の同期コードを async/await に安全に書き換えてください。
## 元コード (同期版)
```python
<同期コードを貼る>
```
## 周辺情報
- Python バージョン: <例: 3.11>
- 使っているライブラリ: <例: requests, sqlalchemy 2.0, redis>
- 並行度の目標: <例: 同時 50 リクエスト>
- 実行環境: <FastAPI / CLI / バックグラウンドワーカー>
## リファクタ方針 (この順で検討)
1. **真の async に置き換えるべき I/O** を列挙
- HTTP: requests → httpx.AsyncClient
- DB: SQLAlchemy → async session / asyncpg
- Redis: redis-py → redis.asyncio
- ファイル: open() → aiofiles (必要な時だけ)
2. **CPU バウンドな処理** は async にせず、run_in_executor / ProcessPoolExecutor で外に出す
3. **同期しか無いライブラリ** は asyncio.to_thread() で逃がす (例: boto3 一部、PIL)
4. **イベントループのブロック検出** (time.sleep / requests.get の残骸が無いか)
5. **並行度の制御** (asyncio.Semaphore / asyncio.gather の chunk 分割)
6. **例外の握りつぶし防止** (gather の return_exceptions=True の扱い)
## 出力
1. **置換マッピング表** (元ライブラリ / async版 / 注意点 の3列)
2. **書き換え後のコード** (型ヒント完備、PEP 8 準拠)
3. **イベントループブロックの危険箇所** (リファクタ後も残るもの)
4. **並行度の推奨値** (Semaphore の数値と根拠)
5. **計測コード** (asyncio.run の前後で時間を測る最小スニペット)
6. **テスト戦略** (pytest-asyncio の最小例)
## 制約
- 同期関数に async def を付けただけの「偽 async」は禁止 (本当に await できるか確認)
- bare except / 例外無視は禁止
- グローバル状態をイベントループ間で共有しない (FastAPI の Depends 等で持ち回す)
効くポイント
- 「真の async」と「executor 逃がし」を最初に分類させると、偽 async リファクタを防げる
- 並行度推奨値と計測コードを要求すると、書き換え後の効果検証まで一気通貫になる
- 置換マッピング表を出させると、他ファイルへの横展開が機械的にできる
よくある質問
- FastAPI ではなく Django や Flask の同期コードでも使えますか?
- リファクタ方針の 1-6 (真の async に置き換えるべき I/O・CPU バウンド処理の executor 逃がし・偽 async 検出・並行度制御) はフレームワーク非依存です。ただし Django は同期 ORM が前提のため、async 化のメリットが薄い場面もあります。プロンプトの『実行環境』欄を Django にした上で、AI に「async 化のメリットがコストを上回るか」を併せて判断させると過剰リファクタを避けられます。
- 並行度の目標値 (同時 N リクエスト) が分かりません
- 「未定」と書いて投げて構いません。AI が一般的な目安として 10-50 の Semaphore を提案してきますので、そこから本番計測で調整する流れで十分です。むしろ最初から具体的な目標値を入れて投げると、その数字に AI が引っ張られて推奨値の柔軟性が落ちることがあります。
- リファクタ後も time.sleep が消えません
- AI が同期コードを残してしまった場合は、出力された書き換え後コードに対して「ripgrep で time.sleep / requests.get / 同期 sqlalchemy session を検索した結果、残骸が無いことを確認してください」と追加で投げると、自己検証してくれます。プロンプト末尾の制約に「偽 async は禁止」と明記している効果はここで初めて発動します。
- asyncio.to_thread で逃がすか、別ライブラリの async 版に置き換えるかの判断基準は?
- AI に「対象ライブラリの async 公式サポートが安定版で存在し、かつ書き換え工数が 1 日以内なら async 版に置換、それ以外は to_thread で逃がす」と判断基準を渡すと一貫した提案が返ります。boto3 のようにメンテナがある分野は aioboto3 がありますが、ライブラリ全体の挙動が変わるリスクもあるので、影響範囲を見て決めるのが安全です。
このプロンプトを実戦で使った所感や改善案があればぜひフィードバックを。姉妹サイト ai-pick.tech では AI x SNS集客の運用ノウハウを公開しています。