Verified against Claude Code · 2026-07-24
Refactor blocking Python code to async without breaking it silently
A prompt for converting synchronous I/O-bound code to async/await that separates what genuinely benefits from async, wraps unavoidable blocking calls correctly, and checks every caller — instead of prefixing every function with async def and calling it done.
The prompt
Ready to copy — highlighted parts are example details you can swap.
Refactor the code below from synchronous to async — but only where it actually matters. Do not prefix every function with async def reflexively; that adds overhead without benefit for CPU-bound code and creates a false sense of concurrency that doesn't actually exist. CODE def fetch_all_prices(product_ids: list[str]) -> list[Price]: loops and calls requests.get() per id I/O OPERATIONS INVOLVED HTTP GET requests to a pricing API, a write to a local CSV file GOAL fetch prices for up to 50 products concurrently instead of one request at a time DEPLOYMENT CONTEXT Runs inside a FastAPI request handler, which already has its own running event loop. REQUIREMENTS 1. For each I/O operation, identify whether an async-native library exists (httpx's AsyncClient instead of requests, an async database driver instead of a sync one, aiofiles for file I/O) and use it. If no async equivalent exists for something in HTTP GET requests to a pricing API, a write to a local CSV file, say so explicitly and wrap it with asyncio.to_thread rather than calling it directly inside an async def and silently blocking the event loop. 2. Any genuinely CPU-bound work (parsing, computation) stays synchronous, or is offloaded to asyncio.to_thread or a ProcessPoolExecutor if it's heavy enough to matter — making it async def alone does nothing for CPU-bound code, and pretending otherwise is the most common mistake in this kind of refactor. 3. Use asyncio.gather, or a TaskGroup on Python 3.11+, to run independent I/O calls concurrently where fetch prices for up to 50 products concurrently instead of one request at a time calls for it — don't await a list of calls sequentially in a loop and call that async. 4. Propagate cancellation and timeouts correctly: wrap calls that should have a deadline in asyncio.timeout(...), and don't swallow asyncio.CancelledError with a bare except Exception, since that breaks cooperative cancellation for whatever is awaiting this code. 5. Confirm Runs inside a FastAPI request handler, which already has its own running event loop. actually supports the change — code running inside an existing event loop (a web framework request handler) needs different treatment than a standalone script that will call asyncio.run() itself, and mixing the two incorrectly is a common source of "RuntimeError: this event loop is already running." 6. Every function whose signature changes from sync to async must have every one of its callers updated too — list them, don't leave a caller doing result = my_func() on a coroutine it never awaits. 7. If any resource used inside the refactored code (a database connection, an HTTP client session) is currently created once and reused across calls, preserve that lifecycle in the async version too — don't silently open a new httpx.AsyncClient per call inside a hot loop just because the sync version's requests.Session was easy to drop, since that trades one performance problem for a worse one. OUTPUT FORMAT 1. The refactored code. 2. A table: function name, sync or async now, and the reason. 3. Every caller found that also needed updating, and confirmation each one now awaits correctly. 4. One line confirming any shared resource's lifecycle (created once vs. per call) matches what the original synchronous code did, or naming why it deliberately changed.
Customize
Optional — swap in your own details for the highlighted parts above.
Why this works
The explicit instruction against reflexively adding async def everywhere addresses a real, common misconception: async does nothing for CPU-bound work and adds event-loop scheduling overhead for no benefit, so a refactor that async-ifies indiscriminately can make code slower and harder to reason about while looking more modern on the surface. Naming asyncio.to_thread for any I/O operation with no async-native library available solves the single most common asyncio production bug directly — a blocking call left inside an async def function, which stalls the entire event loop for every concurrent request being served by that worker, not just the one making the slow call, and a load test with a single client will never reveal this because there's nothing else in the loop for it to block. The deployment_context field exists because the correct top-level pattern genuinely differs by where the code runs: a standalone script owns its own event loop via asyncio.run(), while code inside a FastAPI or similar handler is already running inside someone else's loop, and calling asyncio.run() again inside that context raises "RuntimeError: this event loop is already running" — a refactor that ignores this distinction produces code that works in isolated testing and breaks the moment it's actually deployed. Requiring the caller list is what catches the failure mode that's easy to miss and doesn't raise an exception when it happens: calling a newly-async function without awaiting it produces a coroutine object and a RuntimeWarning, not a crash, so the code silently does nothing useful at that call site unless someone specifically checks for the warning or notices the missing side effect downstream. Preserving a shared resource's create-once lifecycle matters for a related reason specific to async HTTP clients: httpx.AsyncClient holds a connection pool, and creating a fresh one inside a hot loop means every single call pays the cost of a new TCP handshake and TLS negotiation instead of reusing a warm connection, which can make an "async" refactor measurably slower under real concurrent load than the synchronous version it replaced, despite looking like a strict improvement in the code.
What you get back
async def fetch_all_prices(product_ids: list[str]) -> list[Price]: async with httpx.AsyncClient() as client: results = await asyncio.gather(*(fetch_price(client, pid) for pid in product_ids)) return results Table: fetch_price -> async (network I/O, httpx.AsyncClient); fetch_all_prices -> async (fans out via gather); parse_price_response -> stays sync (pure CPU-bound parsing, no I/O). Deployment: runs inside FastAPI's already-running loop, so this is awaited directly from the route handler — no asyncio.run() call added anywhere in this code. Callers updated: report_generator.build_report() now does "prices = await fetch_all_prices(ids)" instead of a direct call — confirmed build_report() is itself async and runs inside the same request context.
Verified against
Claude Code Sonnet 4.6 · 2026-07-24
Claude Sonnet 4.6 · 2026-07-25
Changelog
- 2026-07-25 — Initial publish, verified against Claude Code and Claude (Sonnet 4.6) on Python 3.12.
Need this built into your business?
If a prompt isn't enough — custom software, built and maintained for you — that's Scult's day job.
EXPLORE CUSTOM SOFTWARE
