Refactor sync+async modules to eliminate duplication using the iter-coroutine
apm install @vercel/iter-coroutine
[](https://apm-p1ls2dz87-atlamors-projects.vercel.app/packages/@vercel/iter-coroutine)# Iter-Coroutine + Base/Runtime Migration
Refactor sync+async modules to eliminate duplication using the iter-coroutine
pattern. This skill covers the full migration workflow: identifying candidates,
structuring internal modules, writing transport-agnostic business logic, and
wiring up sync/async public APIs.
## When to use
- A module has parallel sync and async implementations with duplicated logic.
- You're adding a new feature that needs both sync and async public APIs.
- You're refactoring an existing feature to use the shared HTTP transport layer.
## Core principles
- **Public API is stable.** Same exported names, signatures, return types, and
behavior. Migrations are internal refactors, not API changes.
- **Internal core is async-first.** All shared business logic lives in async
methods on a base class.
- **Sync entrypoints are thin wrappers** over the async core via
`iter_coroutine(...)`, used only when the wrapped coroutine is non-suspending
in sync mode.
- **HTTP goes through `vercel._internal.http`** transports. Never construct
`httpx.Client` or `httpx.AsyncClient` directly in feature modules.
## Architecture overview
```
Public API (sync wrappers + async functions)
│
│ iter_coroutine() one-shot bridge (sync path)
│ await (async path)
▼
Concrete Clients (SyncFooClient / AsyncFooClient)
│
│ inherit from
▼
BaseFooClient (all async methods — shared business logic)
│
│ uses
▼
RequestClient (transport-agnostic async API)
│
▼
Transport (SyncTransport or AsyncTransport)
│
▼
httpx.Client or httpx.AsyncClient
```
## How `iter_coroutine` works
```python
# src/vercel/_internal/iter_coroutine.py
def iter_coroutine(coro: Coroutine[None, None, _T]) -> _T:
"""Execute a non-suspending coroutine synchronously."""
try:
coro.send(None)
except StopIteration as ex:
return ex.value
else:
raise RuntimeError(f"coroutine {coro!r} did not stop after one iteration!")
finally:
coro.close()
```
It drives a coroutine forward exactly one step. If the coroutine completes
without suspending, its return value is extracted from the `StopIteration`
exception. If it tries to yield/await, it raises `RuntimeError`.
### Safety rules
| Safe | Unsafe |
|---------------------------------------------------|--------------------------------------------------|
| Coroutines that only call sync code via async API | Coroutines that `await` real I/O |
| `SyncTransport.send()` (sync under async facade) | `AsyncTransport.send()` (real `await`) |
| Sync sleep, sync file I/O | `asyncio.sleep`, `asyncio.create_task` |
| `inspect.isawaitable()` guarded branches | Unguarded `await` on user callbacks |
## Step-by-step migration
### 1. Create the transport-agnostic request client
The request client wraps a `BaseTransport` and exposes an async API. The
transport implementation determines whether I/O is actually sync or async.
```python
# src/vercel/_internal/foo/core.py
from vercel._internal.http import (
AsyncTransport, BaseTransport, SyncTransport,
create_base_client, create_base_async_client,
)
class FooRequestClient:
def __init__(
self,
*,
transport: BaseTransport,
sleep_fn: SleepFn = asyncio.sleep,
await_callback: bool = True,
) -> None:
self._transport = transport
self._sleep_fn = sleep_fn
self._await_callback = await_callback
async def request(self, method: str, path: str, **kwargs) -> Any:
# Retry logic, error mapping, progress callbacks — all async-shaped
# but non-suspending when backed by SyncTransport
resp = await self._transport.send(method, path, **kwargs)
...
```
Factory functions configure transport + sleep + callback behavior:
```python
def create_sync_request_client(timeout: float = 30.0) -> FooRequestClient:
transport = SyncTransport(create_base_client(timeout=timeout))
return FooRequestClient(
transport=transport,
sleep_fn=_sync_sleep, # time.sleep, not asyncio.sleep
await_callback=False, # don't await user callbacks
)
def create_async_request_client(timeout: float = 30.0) -> FooRequestClient:
transport = AsyncTransport(create_base_async_client(timeout=timeout))
return FooRequestClient(transport=transport) # defaults are async-friendly
```
### 2. Write the shared base client
All business logic goes in async methods on a base class:
```python
class BaseFooClient:
def __init__(self, *, request_client: FooRequestClient) -> None:
self._request_client = request_client
async def do_thing(self, path: str, *, token: str) -> Result:
# Validation, request building, response parsing — all here
raw = await self._request_client.request("POST", path, token=token)
return build_result(raw)
```
### 3. Add sync and async concrete clients
```python
class SyncFooClient(BaseFooClient):
def __init__(self) -> None:
super().__init__(request_client=create_sync_request_client())
def close(self) -> None:
self._request_client.close()
def __enter__(self) -> SyncFooClient:
return self
def __exit__(self, *args: object) -> None:
self.close()
class AsyncFooClient(BaseFooClient):
def __init__(self) -> None:
super().__init__(request_client=create_async_request_client())
async def aclose(self) -> None:
await self._request_client.aclose()
async def __aenter__(self) -> AsyncFooClient:
return self
async def __aexit__(self, *args: object) -> None:
await self.aclose()
```
### 4. Wire up public API functions
```python
# src/vercel/foo/ops.py
from vercel._internal.iter_coroutine import iter_coroutine
_T = TypeVar("_T")
def _run_sync(
operation: Callable[[SyncFooClient], Coroutine[None, None, _T]],
) -> _T:
with SyncFooClient() as client:
return iter_coroutine(operation(client))
# Sync public API
def do_thing(path: str, *, token: str | None = None) -> Result:
token = ensure_token(token)
return _run_sync(lambda c: c.do_thing(path, token=token))
# Async public API
async def do_thing_async(path: str, *, token: str | None = None) -> Result:
token = ensure_token(token)
async with AsyncFooClient() as client:
return await client.do_thing(path, token=token)
```
## Base + runtime split
When sync and async paths differ materially (e.g., threading vs asyncio for
concurrent uploads), use a runtime object:
```python
# Sync runtime — ThreadPoolExecutor
class _SyncUploadRuntime:
def upload(self, *, parts, upload_fn, ...) -> list[PartResult]:
with ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as executor:
# Submit work, collect results
...
# Async runtime — anyio task group
class _AsyncUploadRuntime:
async def upload(self, *, parts, upload_fn, ...) -> list[PartResult]:
semaphore = anyio.Semaphore(MAX_CONCURRENCY)
async with anyio.create_task_group() as tg:
# Start tasks with semaphore limiting
...
```
The base client calls the runtime polymorphically:
```python
class BaseFooClient:
async def _do_upload(self, ...) -> list[PartResult]:
parts = await _await_if_necessary(
self._runtime.upload(
parts=chunks,
upload_fn=self._make_upload_fn(),
...
)
)
return order_parts(parts)
```
Concrete clients inject the right runtime and upload function:
```python
class SyncFooClient(BaseFooClient):
def __init__(self) -> None:
super().__init__(runtime=create_sync_runtime())
def _make_upload_fn(self):
# Wrap async method with iter_coroutine for sync execution
return lambda **kw: iter_coroutine(self._client.upload_part(**kw))
class AsyncFooClient(BaseFooClient):
def __init__(self) -> None:
super().__init__(runtime=create_async_runtime())
def _make_upload_fn(self):
# Return the async method directly
return self._client.upload_part
```
### Shared orchestration
Keep these in the base layer, not duplicated per runtime:
- Input validation
- Chunk/part iteration helpers
- Result normalization and ordering
- Final response shaping
## Callback handling
For callbacks that may be sync or async:
```python
async def _emit_progress(
callback: Callable[[Event], None] | Callable[[Event], Awaitable[None]] | None,
event: Event,
*,
await_callback: bool,
) -> None:
if callback is None:
return
result = callback(event)
if await_callback and inspect.isawaitable(result):
await cast(Awaitable[None], result)
```
- Sync clients set `await_callback=False` — callbacks are always sync.
- Async clients set `await_callback=True` — callbacks may be awaitable.
## Testing expectations
- Prefer integration-style tests with `respx` that verify real request flow and
sync/async parity.
- Do not rely only on monkeypatch tests that assert internal call shape.
- Test both sync and async paths for every operation.
- Validate before commit:
- `./scripts/lint.sh`
- `./scripts/test.sh` (targeted or full)
## Reference implementation
The blob module is the canonical example of this pattern:
- **Transport:** `src/vercel/_internal/http/transport.py`
- **Request client:** `src/vercel/_internal/blob/core.py` → `BlobRequestClient`
- **Base client:** `src/vercel/_internal/blob/core.py` → `BaseBlobOpsClient`
- **Sync client:** `src/vercel/_internal/blob/core.py` → `SyncBlobOpsClient`
- **Async client:** `src/vercel/_internal/blob/core.py` → `AsyncBlobOpsClient`
- **Runtime split:** `src/vercel/_internal/blob/multipart.py`
- **Public API:** `src/vercel/blob/ops.py`
- **Types re-export:** `src/vercel/blob/types.py`