Channel

Recording channel wrappers that intercept gRPC calls for recording and playback.

Overview

grpcvcr provides channel wrappers that intercept all gRPC calls, recording them during the first run and replaying them from the cassette on subsequent runs.

from grpcvcr import RecordMode, recorded_channel

# Simple usage with context manager
with recorded_channel("test.yaml", "localhost:50051", record_mode=RecordMode.ALL) as channel:
    stub = MyServiceStub(channel)
    response = stub.GetUser(GetUserRequest(id=1))

API Reference

A gRPC channel wrapper that records and plays back interactions.

Wraps a standard gRPC channel with interceptors that record all RPC calls to a cassette for later playback.

Example

from grpcvcr import Cassette, RecordingChannel, RecordMode

cassette = Cassette("tests/cassettes/test.yaml", record_mode=RecordMode.ALL)

with RecordingChannel(cassette, "localhost:50051") as recording:
    stub = MyServiceStub(recording.channel)
    response = stub.GetUser(GetUserRequest(id=1))

Parameters:

Name	Type	Description	Default
cassette	Cassette	The cassette to record to or play back from.	required
target	str	The gRPC server address (e.g., 'localhost:50051').	required
credentials	ChannelCredentials | None	Optional channel credentials for secure channels.	None
options	list[tuple[str, str]] | None	Optional gRPC channel options.	None

Source code in src/grpcvcr/channel.py

def __init__(
    self,
    cassette: Cassette,
    target: str,
    *,
    credentials: grpc.ChannelCredentials | None = None,
    options: list[tuple[str, str]] | None = None,
) -> None:
    """Create a new RecordingChannel.

    Args:
        cassette: The cassette to record to or play back from.
        target: The gRPC server address (e.g., 'localhost:50051').
        credentials: Optional channel credentials for secure channels.
        options: Optional gRPC channel options.
    """
    self.cassette = cassette
    """The cassette being used for recording/playback."""

    self.target = target
    """The gRPC server address."""

    interceptors = create_interceptors(cassette)

    if credentials:
        base_channel = grpc.secure_channel(target, credentials, options=options or [])
    else:
        base_channel = grpc.insecure_channel(target, options=options or [])

    self.channel = grpc.intercept_channel(base_channel, *interceptors)
    """The wrapped gRPC channel with recording interceptors."""

channel instance-attribute

channel = intercept_channel(base_channel, *interceptors)

The wrapped gRPC channel with recording interceptors.

__enter__

__enter__() -> RecordingChannel

Source code in src/grpcvcr/channel.py

def __enter__(self) -> RecordingChannel:
    return self

__exit__

__exit__(*args: object) -> None

Source code in src/grpcvcr/channel.py

def __exit__(self, *args: object) -> None:
    self.close()

Context manager for a recorded gRPC channel.

This is the simplest API for recording and playing back gRPC calls. Creates a cassette and channel, yields the channel, then saves the cassette on exit.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the cassette file.	required
target	str	The gRPC server address.	required
record_mode	RecordMode	How to handle recording vs playback.	NEW_EPISODES
match_on	Matcher | None	Matcher to use for finding recorded interactions.	None
credentials	ChannelCredentials | None	Optional channel credentials for secure channels.	None
options	list[tuple[str, str]] | None	Optional gRPC channel options.	None

Yields:

Type	Description
Channel	A gRPC channel that records/plays back interactions.

Example

from grpcvcr import recorded_channel, RecordMode

# Record interactions
with recorded_channel("test.yaml", "localhost:50051", record_mode=RecordMode.ALL) as channel:
    stub = MyServiceStub(channel)
    response = stub.GetUser(GetUserRequest(id=1))

# Play back interactions (no network calls made)
with recorded_channel("test.yaml", "localhost:50051", record_mode=RecordMode.NONE) as channel:
    stub = MyServiceStub(channel)
    response = stub.GetUser(GetUserRequest(id=1))

Source code in src/grpcvcr/channel.py

@contextmanager
def recorded_channel(
    path: str | Path,
    target: str,
    *,
    record_mode: RecordMode = RecordMode.NEW_EPISODES,
    match_on: Matcher | None = None,
    credentials: grpc.ChannelCredentials | None = None,
    options: list[tuple[str, str]] | None = None,
) -> Generator[grpc.Channel, None, None]:
    """Context manager for a recorded gRPC channel.

    This is the simplest API for recording and playing back gRPC calls.
    Creates a cassette and channel, yields the channel, then saves the
    cassette on exit.

    Args:
        path: Path to the cassette file.
        target: The gRPC server address.
        record_mode: How to handle recording vs playback.
        match_on: Matcher to use for finding recorded interactions.
        credentials: Optional channel credentials for secure channels.
        options: Optional gRPC channel options.

    Yields:
        A gRPC channel that records/plays back interactions.

    Example:
        ```python
        from grpcvcr import recorded_channel, RecordMode

        # Record interactions
        with recorded_channel("test.yaml", "localhost:50051", record_mode=RecordMode.ALL) as channel:
            stub = MyServiceStub(channel)
            response = stub.GetUser(GetUserRequest(id=1))

        # Play back interactions (no network calls made)
        with recorded_channel("test.yaml", "localhost:50051", record_mode=RecordMode.NONE) as channel:
            stub = MyServiceStub(channel)
            response = stub.GetUser(GetUserRequest(id=1))
        ```
    """
    cassette = Cassette(
        path=Path(path),
        record_mode=record_mode,
        match_on=match_on or DEFAULT_MATCHER,
    )

    recording = RecordingChannel(cassette, target, credentials=credentials, options=options)

    try:
        yield recording.channel
    finally:
        recording.close()