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

# 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

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.

# 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:

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:

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

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

# 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:

# 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:

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:

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:

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