@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`