Skip to content

API Reference

This page documents the public API of the interposition library.

Core Module

interposition

Protocol-agnostic interaction interposition with lifecycle hooks.

Provides record, replay, and control capabilities.

BrokerMode module-attribute 

BrokerMode = Literal['replay', 'record', 'auto']

Broker

Manages interaction replay from cassettes.

Attributes:

Name Type Description
cassette Cassette

The cassette containing recorded interactions
mode BrokerMode

The broker mode (replay, record, or auto)
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto; optional in replay mode.
cassette_store CassetteStore | None

Optional store for cassette persistence
Source code in src/interposition/services.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
class Broker:
    """Manages interaction replay from cassettes.

    Attributes:
        cassette: The cassette containing recorded interactions
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when mode is
            record or auto; optional in replay mode.
        cassette_store: Optional store for cassette persistence
    """

    def __init__(
        self,
        cassette: Cassette,
        mode: BrokerMode = "replay",
        live_responder: LiveResponder | None = None,
        cassette_store: CassetteStore | None = None,
    ) -> None:
        """Initialize broker with a cassette.

        Args:
            cassette: The cassette containing recorded interactions
            mode: The broker mode (replay, record, or auto)
            live_responder: Callable for upstream forwarding. Required when
                mode is record or auto.
            cassette_store: Optional store for automatic cassette persistence

        Raises:
            LiveResponderRequiredError: When mode is record or auto and
                live_responder is not configured.
        """
        if mode in ("record", "auto") and live_responder is None:
            raise LiveResponderRequiredError(mode)

        self._cassette = cassette
        self._mode = mode
        self._live_responder = live_responder
        self._cassette_store = cassette_store

    @classmethod
    def from_store(
        cls,
        cassette_store: CassetteStore,
        mode: BrokerMode = "replay",
        live_responder: LiveResponder | None = None,
    ) -> Self:
        """Create a Broker by loading a cassette from a store.

        Args:
            cassette_store: The store to load the cassette from
            mode: The broker mode (replay, record, or auto)
            live_responder: Callable for upstream forwarding. Required when
                mode is record or auto.

        Returns:
            A new Broker instance with the loaded cassette.

        Raises:
            CassetteLoadError: When the store fails to load the cassette
                (e.g., missing file, corrupted data).
            LiveResponderRequiredError: When mode is record or auto and
                live_responder is not configured.
        """
        cassette = cassette_store.load()
        return cls(
            cassette=cassette,
            mode=mode,
            live_responder=live_responder,
            cassette_store=cassette_store,
        )

    @property
    def cassette(self) -> Cassette:
        """Get the cassette."""
        return self._cassette

    @property
    def mode(self) -> BrokerMode:
        """Get the broker mode."""
        return self._mode

    @property
    def live_responder(self) -> LiveResponder | None:
        """Get the live responder."""
        return self._live_responder

    @property
    def cassette_store(self) -> CassetteStore | None:
        """Get the cassette store."""
        return self._cassette_store

    def replay(self, request: InteractionRequest) -> Iterator[ResponseChunk]:
        """Replay recorded response for matching request.

        Args:
            request: The request to match and replay

        Yields:
            ResponseChunks in original recorded order

        Raises:
            InteractionNotFoundError: When no matching interaction exists
                and mode is replay.
        """
        # record mode: always forward to live, ignore cassette
        if self._mode == "record":
            yield from self._forward_and_record(request)
            return

        # replay/auto mode: try cassette first
        interaction = self.cassette.find_interaction(request.fingerprint())
        if interaction is not None:
            yield from interaction.response_chunks
            return

        # MISS handling
        if self._mode == "replay":
            raise InteractionNotFoundError(request)

        # auto mode MISS: forward to live
        yield from self._forward_and_record(request)

    def _forward_and_record(
        self, request: InteractionRequest
    ) -> Iterator[ResponseChunk]:
        """Forward request to live responder and record the interaction.

        Args:
            request: The request to forward

        Yields:
            ResponseChunks from live responder

        Raises:
            LiveResponderRequiredError: When live_responder is not configured.
        """
        if self._live_responder is None:
            raise LiveResponderRequiredError(self._mode)

        chunks = tuple(self._live_responder(request))
        self._record_interaction(request, chunks)
        if self._cassette_store is not None:
            self._cassette_store.save(self._cassette)
        yield from chunks

    def _record_interaction(
        self,
        request: InteractionRequest,
        response_chunks: tuple[ResponseChunk, ...],
    ) -> None:
        """Record a new interaction to the cassette.

        Creates a new Cassette with the interaction appended.

        Args:
            request: The request that was made
            response_chunks: The response chunks from live responder
        """
        interaction = Interaction(
            request=request,
            fingerprint=request.fingerprint(),
            response_chunks=response_chunks,
        )
        new_interactions = (*self._cassette.interactions, interaction)
        self._cassette = Cassette(interactions=new_interactions)

cassette property 

cassette: Cassette

Get the cassette.

mode property 

mode: BrokerMode

Get the broker mode.

live_responder property 

live_responder: LiveResponder | None

Get the live responder.

cassette_store property 

cassette_store: CassetteStore | None

Get the cassette store.

__init__ 

__init__(
    cassette: Cassette,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
    cassette_store: CassetteStore | None = None,
) -> None

Initialize broker with a cassette.

Parameters:

Name Type Description Default
cassette Cassette

The cassette containing recorded interactions

 required
mode BrokerMode

The broker mode (replay, record, or auto)

 'replay'
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto.

 None
cassette_store CassetteStore | None

Optional store for automatic cassette persistence

 None

Raises:

Type Description
LiveResponderRequiredError

When mode is record or auto and live_responder is not configured.
Source code in src/interposition/services.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    cassette: Cassette,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
    cassette_store: CassetteStore | None = None,
) -> None:
    """Initialize broker with a cassette.

    Args:
        cassette: The cassette containing recorded interactions
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when
            mode is record or auto.
        cassette_store: Optional store for automatic cassette persistence

    Raises:
        LiveResponderRequiredError: When mode is record or auto and
            live_responder is not configured.
    """
    if mode in ("record", "auto") and live_responder is None:
        raise LiveResponderRequiredError(mode)

    self._cassette = cassette
    self._mode = mode
    self._live_responder = live_responder
    self._cassette_store = cassette_store

from_store classmethod 

from_store(
    cassette_store: CassetteStore,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
) -> Self

Create a Broker by loading a cassette from a store.

Parameters:

Name Type Description Default
cassette_store CassetteStore

The store to load the cassette from

 required
mode BrokerMode

The broker mode (replay, record, or auto)

 'replay'
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto.

 None

Returns:

Type Description
Self

A new Broker instance with the loaded cassette.

Raises:

Type Description
CassetteLoadError

When the store fails to load the cassette (e.g., missing file, corrupted data).
LiveResponderRequiredError

When mode is record or auto and live_responder is not configured.
Source code in src/interposition/services.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def from_store(
    cls,
    cassette_store: CassetteStore,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
) -> Self:
    """Create a Broker by loading a cassette from a store.

    Args:
        cassette_store: The store to load the cassette from
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when
            mode is record or auto.

    Returns:
        A new Broker instance with the loaded cassette.

    Raises:
        CassetteLoadError: When the store fails to load the cassette
            (e.g., missing file, corrupted data).
        LiveResponderRequiredError: When mode is record or auto and
            live_responder is not configured.
    """
    cassette = cassette_store.load()
    return cls(
        cassette=cassette,
        mode=mode,
        live_responder=live_responder,
        cassette_store=cassette_store,
    )

replay 

replay(
    request: InteractionRequest,
) -> Iterator[ResponseChunk]

Replay recorded response for matching request.

Parameters:

Name Type Description Default
request InteractionRequest

The request to match and replay

 required

Yields:

Type Description
ResponseChunk

ResponseChunks in original recorded order

Raises:

Type Description
InteractionNotFoundError

When no matching interaction exists and mode is replay.
Source code in src/interposition/services.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def replay(self, request: InteractionRequest) -> Iterator[ResponseChunk]:
    """Replay recorded response for matching request.

    Args:
        request: The request to match and replay

    Yields:
        ResponseChunks in original recorded order

    Raises:
        InteractionNotFoundError: When no matching interaction exists
            and mode is replay.
    """
    # record mode: always forward to live, ignore cassette
    if self._mode == "record":
        yield from self._forward_and_record(request)
        return

    # replay/auto mode: try cassette first
    interaction = self.cassette.find_interaction(request.fingerprint())
    if interaction is not None:
        yield from interaction.response_chunks
        return

    # MISS handling
    if self._mode == "replay":
        raise InteractionNotFoundError(request)

    # auto mode MISS: forward to live
    yield from self._forward_and_record(request)

Cassette

Bases: BaseModel

In-memory collection of recorded interactions.

Attributes:

Name Type Description
interactions tuple[Interaction, ...]

Ordered sequence of interactions
Source code in src/interposition/models.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class Cassette(BaseModel):
    """In-memory collection of recorded interactions.

    Attributes:
        interactions: Ordered sequence of interactions
    """

    model_config = ConfigDict(frozen=True)

    interactions: tuple[Interaction, ...]
    _index: dict[RequestFingerprint, int] = PrivateAttr(default_factory=dict)

    @model_validator(mode="after")
    def build_index(self) -> Self:
        """Build fingerprint index for efficient lookup."""
        index: dict[RequestFingerprint, int] = {}
        for i, interaction in enumerate(self.interactions):
            # Only store first occurrence of each fingerprint
            if interaction.fingerprint not in index:
                index[interaction.fingerprint] = i
        # Use object.__setattr__ to modify frozen model
        object.__setattr__(self, "_index", index)
        return self

    def find_interaction(self, fingerprint: RequestFingerprint) -> Interaction | None:
        """Find first interaction matching fingerprint.

        Args:
            fingerprint: Request fingerprint to search for

        Returns:
            Matching Interaction or None if not found
        """
        position = self._index.get(fingerprint)
        if position is None:
            return None
        return self.interactions[position]

build_index 

build_index() -> Self

Build fingerprint index for efficient lookup.

Source code in src/interposition/models.py 
203
204
205
206
207
208
209
210
211
212
213
@model_validator(mode="after")
def build_index(self) -> Self:
    """Build fingerprint index for efficient lookup."""
    index: dict[RequestFingerprint, int] = {}
    for i, interaction in enumerate(self.interactions):
        # Only store first occurrence of each fingerprint
        if interaction.fingerprint not in index:
            index[interaction.fingerprint] = i
    # Use object.__setattr__ to modify frozen model
    object.__setattr__(self, "_index", index)
    return self

find_interaction 

find_interaction(
    fingerprint: RequestFingerprint,
) -> Interaction | None

Find first interaction matching fingerprint.

Parameters:

Name Type Description Default
fingerprint RequestFingerprint

Request fingerprint to search for

 required

Returns:

Type Description
Interaction | None

Matching Interaction or None if not found
Source code in src/interposition/models.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
def find_interaction(self, fingerprint: RequestFingerprint) -> Interaction | None:
    """Find first interaction matching fingerprint.

    Args:
        fingerprint: Request fingerprint to search for

    Returns:
        Matching Interaction or None if not found
    """
    position = self._index.get(fingerprint)
    if position is None:
        return None
    return self.interactions[position]

CassetteStore

Bases: Protocol

Port for cassette persistence operations.

Implementations handle loading and saving cassettes to storage. The Broker calls save() automatically after recording new interactions.

Source code in src/interposition/services.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class CassetteStore(Protocol):
    """Port for cassette persistence operations.

    Implementations handle loading and saving cassettes to storage.
    The Broker calls save() automatically after recording new interactions.
    """

    def load(self) -> Cassette:
        """Load cassette from storage.

        Returns:
            The loaded Cassette instance.
        """
        ...

    def save(self, cassette: Cassette) -> None:
        """Save cassette to storage.

        Args:
            cassette: The cassette to persist.
        """
        ...

load 

load() -> Cassette

Load cassette from storage.

Returns:

Type Description
Cassette

The loaded Cassette instance.
Source code in src/interposition/services.py 
29
30
31
32
33
34
35
def load(self) -> Cassette:
    """Load cassette from storage.

    Returns:
        The loaded Cassette instance.
    """
    ...

save 

save(cassette: Cassette) -> None

Save cassette to storage.

Parameters:

Name Type Description Default
cassette Cassette

The cassette to persist.

 required
Source code in src/interposition/services.py 
37
38
39
40
41
42
43
def save(self, cassette: Cassette) -> None:
    """Save cassette to storage.

    Args:
        cassette: The cassette to persist.
    """
    ...

Interaction

Bases: BaseModel

Complete request-response pair for replay.

Attributes:

Name Type Description
request InteractionRequest

The original InteractionRequest
fingerprint RequestFingerprint

Precomputed request fingerprint for matching
response_chunks tuple[ResponseChunk, ...]

Ordered sequence of response chunks
metadata tuple[tuple[str, str], ...]

Optional interaction metadata as (key, value) pairs. Examples: recording timestamp, session ID, test scenario name. Useful for debugging and tracing recorded interactions. Default is empty tuple.
Source code in src/interposition/models.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class Interaction(BaseModel):
    """Complete request-response pair for replay.

    Attributes:
        request: The original InteractionRequest
        fingerprint: Precomputed request fingerprint for matching
        response_chunks: Ordered sequence of response chunks
        metadata: Optional interaction metadata as (key, value) pairs.
            Examples: recording timestamp, session ID, test scenario name.
            Useful for debugging and tracing recorded interactions.
            Default is empty tuple.
    """

    model_config = ConfigDict(frozen=True)

    request: InteractionRequest
    fingerprint: RequestFingerprint
    response_chunks: tuple[ResponseChunk, ...]
    metadata: tuple[tuple[str, str], ...] = ()

    @model_validator(mode="after")
    def validate_interaction(self) -> Self:
        """Validate interaction integrity.

        Raises:
            InteractionValidationError: If fingerprint doesn't match request
                or chunks aren't sequential
        """
        # Verify fingerprint matches request
        expected_fingerprint = self.request.fingerprint()
        if self.fingerprint != expected_fingerprint:
            msg = (
                f"Fingerprint does not match request: "
                f"expected {expected_fingerprint.value}, got {self.fingerprint.value}"
            )
            raise InteractionValidationError(msg)

        # Verify response chunks are sequentially ordered
        if not self.response_chunks:
            msg = "Response chunks cannot be empty"
            raise InteractionValidationError(msg)

        if self.response_chunks[0].sequence != 0:
            msg = "Response chunks must start at sequence 0"
            raise InteractionValidationError(msg)

        for i, chunk in enumerate(self.response_chunks):
            if chunk.sequence != i:
                msg = "Response chunks must be sequential with no gaps"
                raise InteractionValidationError(msg)

        return self

validate_interaction 

validate_interaction() -> Self

Validate interaction integrity.

Raises:

Type Description
InteractionValidationError

If fingerprint doesn't match request or chunks aren't sequential
Source code in src/interposition/models.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@model_validator(mode="after")
def validate_interaction(self) -> Self:
    """Validate interaction integrity.

    Raises:
        InteractionValidationError: If fingerprint doesn't match request
            or chunks aren't sequential
    """
    # Verify fingerprint matches request
    expected_fingerprint = self.request.fingerprint()
    if self.fingerprint != expected_fingerprint:
        msg = (
            f"Fingerprint does not match request: "
            f"expected {expected_fingerprint.value}, got {self.fingerprint.value}"
        )
        raise InteractionValidationError(msg)

    # Verify response chunks are sequentially ordered
    if not self.response_chunks:
        msg = "Response chunks cannot be empty"
        raise InteractionValidationError(msg)

    if self.response_chunks[0].sequence != 0:
        msg = "Response chunks must start at sequence 0"
        raise InteractionValidationError(msg)

    for i, chunk in enumerate(self.response_chunks):
        if chunk.sequence != i:
            msg = "Response chunks must be sequential with no gaps"
            raise InteractionValidationError(msg)

    return self

InteractionRequest

Bases: BaseModel

Structured representation of a protocol-agnostic request.

Attributes:

Name Type Description
protocol str

Protocol identifier (e.g., "grpc", "graphql", "mqtt")
action str

Action/method name (e.g., "ListUsers", "query", "publish")
target str

Target resource (e.g., "users.UserService", "topic/sensors")
headers tuple[tuple[str, str], ...]

Request headers as immutable sequence of key-value pairs
body bytes

Request body content as bytes
Source code in src/interposition/models.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class InteractionRequest(BaseModel):
    """Structured representation of a protocol-agnostic request.

    Attributes:
        protocol: Protocol identifier (e.g., "grpc", "graphql", "mqtt")
        action: Action/method name (e.g., "ListUsers", "query", "publish")
        target: Target resource (e.g., "users.UserService", "topic/sensors")
        headers: Request headers as immutable sequence of key-value pairs
        body: Request body content as bytes
    """

    model_config = ConfigDict(frozen=True)

    protocol: str
    action: str
    target: str
    headers: tuple[tuple[str, str], ...]
    body: bytes

    def fingerprint(self) -> RequestFingerprint:
        """Generate stable fingerprint for efficient matching.

        Returns:
            RequestFingerprint derived from all request fields.
        """
        return RequestFingerprint.from_request(self)

fingerprint 

fingerprint() -> RequestFingerprint

Generate stable fingerprint for efficient matching.

Returns:

Type Description
RequestFingerprint

RequestFingerprint derived from all request fields.
Source code in src/interposition/models.py 
67
68
69
70
71
72
73
def fingerprint(self) -> RequestFingerprint:
    """Generate stable fingerprint for efficient matching.

    Returns:
        RequestFingerprint derived from all request fields.
    """
    return RequestFingerprint.from_request(self)

JsonFileCassetteStore

File-based cassette store using JSON format.

Attributes:

Name Type Description
path Path

Path to the JSON file for cassette storage.
Source code in src/interposition/stores.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class JsonFileCassetteStore:
    """File-based cassette store using JSON format.

    Attributes:
        path: Path to the JSON file for cassette storage.
    """

    def __init__(self, path: Path, *, create_if_missing: bool = False) -> None:
        """Initialize store with file path.

        Args:
            path: Path to the JSON file (will be created if doesn't exist).
            create_if_missing: If True, load() returns an empty Cassette
                when the file doesn't exist instead of raising CassetteLoadError.
        """
        self._path = path
        self._create_if_missing = create_if_missing

    @property
    def path(self) -> Path:
        """Get the file path."""
        return self._path

    def load(self) -> Cassette:
        """Load cassette from JSON file.

        Returns:
            Cassette instance loaded from file. If create_if_missing is True
            and the file doesn't exist, returns an empty Cassette.

        Raises:
            CassetteLoadError: If file doesn't exist (and create_if_missing
                is False), file is unreadable, or JSON is invalid.
        """
        try:
            json_str = self._path.read_text(encoding="utf-8")
        except FileNotFoundError as e:
            if self._create_if_missing:
                return Cassette(interactions=())
            raise CassetteLoadError(self._path, e) from e
        except OSError as e:
            raise CassetteLoadError(self._path, e) from e
        try:
            return Cassette.model_validate_json(json_str)
        except Exception as e:
            raise CassetteLoadError(self._path, e) from e

    def save(self, cassette: Cassette) -> None:
        """Save cassette to JSON file.

        Creates parent directories if they don't exist.

        Args:
            cassette: The cassette to persist.

        Raises:
            CassetteSaveError: If file write fails.
        """
        try:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            json_str = cassette.model_dump_json(indent=2)
            self._path.write_text(json_str, encoding="utf-8")
        except OSError as e:
            raise CassetteSaveError(self._path, e) from e

path property 

path: Path

Get the file path.

__init__ 

__init__(
    path: Path, *, create_if_missing: bool = False
) -> None

Initialize store with file path.

Parameters:

Name Type Description Default
path Path

Path to the JSON file (will be created if doesn't exist).

 required
create_if_missing bool

If True, load() returns an empty Cassette when the file doesn't exist instead of raising CassetteLoadError.

 False
Source code in src/interposition/stores.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, path: Path, *, create_if_missing: bool = False) -> None:
    """Initialize store with file path.

    Args:
        path: Path to the JSON file (will be created if doesn't exist).
        create_if_missing: If True, load() returns an empty Cassette
            when the file doesn't exist instead of raising CassetteLoadError.
    """
    self._path = path
    self._create_if_missing = create_if_missing

load 

load() -> Cassette

Load cassette from JSON file.

Returns:

Type Description
Cassette

Cassette instance loaded from file. If create_if_missing is True
Cassette

and the file doesn't exist, returns an empty Cassette.

Raises:

Type Description
CassetteLoadError

If file doesn't exist (and create_if_missing is False), file is unreadable, or JSON is invalid.
Source code in src/interposition/stores.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def load(self) -> Cassette:
    """Load cassette from JSON file.

    Returns:
        Cassette instance loaded from file. If create_if_missing is True
        and the file doesn't exist, returns an empty Cassette.

    Raises:
        CassetteLoadError: If file doesn't exist (and create_if_missing
            is False), file is unreadable, or JSON is invalid.
    """
    try:
        json_str = self._path.read_text(encoding="utf-8")
    except FileNotFoundError as e:
        if self._create_if_missing:
            return Cassette(interactions=())
        raise CassetteLoadError(self._path, e) from e
    except OSError as e:
        raise CassetteLoadError(self._path, e) from e
    try:
        return Cassette.model_validate_json(json_str)
    except Exception as e:
        raise CassetteLoadError(self._path, e) from e

save 

save(cassette: Cassette) -> None

Save cassette to JSON file.

Creates parent directories if they don't exist.

Parameters:

Name Type Description Default
cassette Cassette

The cassette to persist.

 required

Raises:

Type Description
CassetteSaveError

If file write fails.
Source code in src/interposition/stores.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def save(self, cassette: Cassette) -> None:
    """Save cassette to JSON file.

    Creates parent directories if they don't exist.

    Args:
        cassette: The cassette to persist.

    Raises:
        CassetteSaveError: If file write fails.
    """
    try:
        self._path.parent.mkdir(parents=True, exist_ok=True)
        json_str = cassette.model_dump_json(indent=2)
        self._path.write_text(json_str, encoding="utf-8")
    except OSError as e:
        raise CassetteSaveError(self._path, e) from e

ResponseChunk

Bases: BaseModel

Discrete piece of response data.

Attributes:

Name Type Description
data bytes

Chunk payload as bytes
sequence int

Zero-based chunk position in response stream
metadata tuple[tuple[str, str], ...]

Optional chunk metadata as (key, value) string pairs. Examples: timing info, encoding, content-type for this chunk. Default is empty tuple.
Source code in src/interposition/models.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ResponseChunk(BaseModel):
    """Discrete piece of response data.

    Attributes:
        data: Chunk payload as bytes
        sequence: Zero-based chunk position in response stream
        metadata: Optional chunk metadata as (key, value) string pairs.
            Examples: timing info, encoding, content-type for this chunk.
            Default is empty tuple.
    """

    model_config = ConfigDict(frozen=True)

    data: bytes
    sequence: int
    metadata: tuple[tuple[str, str], ...] = ()

RequestFingerprint

Bases: BaseModel

Stable unique identifier for request matching.

Attributes:

Name Type Description
value str

SHA-256 hash of canonicalized request fields
Source code in src/interposition/models.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class RequestFingerprint(BaseModel):
    """Stable unique identifier for request matching.

    Attributes:
        value: SHA-256 hash of canonicalized request fields
    """

    model_config = ConfigDict(frozen=True)

    value: str

    @field_validator("value")
    @classmethod
    def validate_sha256_hex(cls, v: str) -> str:
        """Validate that value is a valid SHA-256 hex string.

        Args:
            v: The fingerprint value to validate

        Returns:
            The validated value

        Raises:
            ValueError: If value is not exactly 64 hex characters
        """
        if len(v) != SHA256_HEX_LENGTH:
            msg = f"SHA-256 hex must be exactly {SHA256_HEX_LENGTH} characters"
            raise ValueError(msg)
        if not all(c in "0123456789abcdef" for c in v):
            msg = "Invalid hex characters in fingerprint"
            raise ValueError(msg)
        return v

    @classmethod
    def from_request(cls, request: InteractionRequest) -> Self:
        """Create fingerprint from InteractionRequest.

        Args:
            request: The request to fingerprint

        Returns:
            RequestFingerprint with SHA-256 hash value
        """
        # Canonical order: protocol, action, target, headers, body
        # Preserve header ordering to avoid normalization.
        canonical_data = [
            request.protocol,
            request.action,
            request.target,
            request.headers,
            request.body.hex(),
        ]
        canonical = json.dumps(
            canonical_data,
            separators=_CANONICAL_JSON_SEPARATORS,
            sort_keys=_CANONICAL_JSON_SORT_KEYS,
        )
        hash_value = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        return cls(value=hash_value)

validate_sha256_hex classmethod 

validate_sha256_hex(v: str) -> str

Validate that value is a valid SHA-256 hex string.

Parameters:

Name Type Description Default
v str

The fingerprint value to validate

 required

Returns:

Type Description
str

The validated value

Raises:

Type Description
ValueError

If value is not exactly 64 hex characters
Source code in src/interposition/models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@field_validator("value")
@classmethod
def validate_sha256_hex(cls, v: str) -> str:
    """Validate that value is a valid SHA-256 hex string.

    Args:
        v: The fingerprint value to validate

    Returns:
        The validated value

    Raises:
        ValueError: If value is not exactly 64 hex characters
    """
    if len(v) != SHA256_HEX_LENGTH:
        msg = f"SHA-256 hex must be exactly {SHA256_HEX_LENGTH} characters"
        raise ValueError(msg)
    if not all(c in "0123456789abcdef" for c in v):
        msg = "Invalid hex characters in fingerprint"
        raise ValueError(msg)
    return v

from_request classmethod 

from_request(request: InteractionRequest) -> Self

Create fingerprint from InteractionRequest.

Parameters:

Name Type Description Default
request InteractionRequest

The request to fingerprint

 required

Returns:

Type Description
Self

RequestFingerprint with SHA-256 hash value
Source code in src/interposition/models.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def from_request(cls, request: InteractionRequest) -> Self:
    """Create fingerprint from InteractionRequest.

    Args:
        request: The request to fingerprint

    Returns:
        RequestFingerprint with SHA-256 hash value
    """
    # Canonical order: protocol, action, target, headers, body
    # Preserve header ordering to avoid normalization.
    canonical_data = [
        request.protocol,
        request.action,
        request.target,
        request.headers,
        request.body.hex(),
    ]
    canonical = json.dumps(
        canonical_data,
        separators=_CANONICAL_JSON_SEPARATORS,
        sort_keys=_CANONICAL_JSON_SORT_KEYS,
    )
    hash_value = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return cls(value=hash_value)

CassetteSaveError

Bases: InterpositionError

Raised when cassette persistence fails.

Source code in src/interposition/errors.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
class CassetteSaveError(InterpositionError):
    """Raised when cassette persistence fails."""

    def __init__(self, path: Path, cause: Exception) -> None:
        """Initialize with the path and underlying cause.

        Args:
            path: The file path where save failed
            cause: The underlying exception that caused the failure
        """
        super().__init__(f"Failed to save cassette to {path}: {cause}")
        self.path: Path = path
        self.__cause__ = cause

__init__ 

__init__(path: Path, cause: Exception) -> None

Initialize with the path and underlying cause.

Parameters:

Name Type Description Default
path Path

The file path where save failed

 required
cause Exception

The underlying exception that caused the failure

 required
Source code in src/interposition/errors.py 
64
65
66
67
68
69
70
71
72
73
def __init__(self, path: Path, cause: Exception) -> None:
    """Initialize with the path and underlying cause.

    Args:
        path: The file path where save failed
        cause: The underlying exception that caused the failure
    """
    super().__init__(f"Failed to save cassette to {path}: {cause}")
    self.path: Path = path
    self.__cause__ = cause

InteractionNotFoundError

Bases: InterpositionError

Raised when no matching interaction is found in cassette.

Source code in src/interposition/errors.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class InteractionNotFoundError(InterpositionError):
    """Raised when no matching interaction is found in cassette."""

    def __init__(self, request: InteractionRequest) -> None:
        """Initialize with request that failed to match.

        Args:
            request: The unmatched request
        """
        super().__init__(
            f"No matching interaction for {request.protocol}:"
            f"{request.action}:{request.target}"
        )
        self.request: InteractionRequest = request

__init__ 

__init__(request: InteractionRequest) -> None

Initialize with request that failed to match.

Parameters:

Name Type Description Default
request InteractionRequest

The unmatched request

 required
Source code in src/interposition/errors.py 
20
21
22
23
24
25
26
27
28
29
30
def __init__(self, request: InteractionRequest) -> None:
    """Initialize with request that failed to match.

    Args:
        request: The unmatched request
    """
    super().__init__(
        f"No matching interaction for {request.protocol}:"
        f"{request.action}:{request.target}"
    )
    self.request: InteractionRequest = request

InteractionValidationError

Bases: InterpositionError, ValueError

Raised when interaction validation fails.

Source code in src/interposition/models.py 
26
27
class InteractionValidationError(InterpositionError, ValueError):
    """Raised when interaction validation fails."""

InterpositionError

Bases: Exception

Base class for all interposition exceptions.

Source code in src/interposition/errors.py 
13
14
class InterpositionError(Exception):
    """Base class for all interposition exceptions."""

LiveResponderRequiredError

Bases: InterpositionError

Raised when live_responder is required but not configured.

Source code in src/interposition/errors.py 
33
34
35
36
37
38
39
40
41
42
43
class LiveResponderRequiredError(InterpositionError):
    """Raised when live_responder is required but not configured."""

    def __init__(self, mode: str) -> None:
        """Initialize with the mode that requires live_responder.

        Args:
            mode: The broker mode that requires live_responder
        """
        super().__init__(f"live_responder is required for {mode} mode")
        self.mode: str = mode

__init__ 

__init__(mode: str) -> None

Initialize with the mode that requires live_responder.

Parameters:

Name Type Description Default
mode str

The broker mode that requires live_responder

 required
Source code in src/interposition/errors.py 
36
37
38
39
40
41
42
43
def __init__(self, mode: str) -> None:
    """Initialize with the mode that requires live_responder.

    Args:
        mode: The broker mode that requires live_responder
    """
    super().__init__(f"live_responder is required for {mode} mode")
    self.mode: str = mode

Services

interposition.services.Broker

Manages interaction replay from cassettes.

Attributes:

Name Type Description
cassette Cassette

The cassette containing recorded interactions
mode BrokerMode

The broker mode (replay, record, or auto)
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto; optional in replay mode.
cassette_store CassetteStore | None

Optional store for cassette persistence
Source code in src/interposition/services.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
class Broker:
    """Manages interaction replay from cassettes.

    Attributes:
        cassette: The cassette containing recorded interactions
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when mode is
            record or auto; optional in replay mode.
        cassette_store: Optional store for cassette persistence
    """

    def __init__(
        self,
        cassette: Cassette,
        mode: BrokerMode = "replay",
        live_responder: LiveResponder | None = None,
        cassette_store: CassetteStore | None = None,
    ) -> None:
        """Initialize broker with a cassette.

        Args:
            cassette: The cassette containing recorded interactions
            mode: The broker mode (replay, record, or auto)
            live_responder: Callable for upstream forwarding. Required when
                mode is record or auto.
            cassette_store: Optional store for automatic cassette persistence

        Raises:
            LiveResponderRequiredError: When mode is record or auto and
                live_responder is not configured.
        """
        if mode in ("record", "auto") and live_responder is None:
            raise LiveResponderRequiredError(mode)

        self._cassette = cassette
        self._mode = mode
        self._live_responder = live_responder
        self._cassette_store = cassette_store

    @classmethod
    def from_store(
        cls,
        cassette_store: CassetteStore,
        mode: BrokerMode = "replay",
        live_responder: LiveResponder | None = None,
    ) -> Self:
        """Create a Broker by loading a cassette from a store.

        Args:
            cassette_store: The store to load the cassette from
            mode: The broker mode (replay, record, or auto)
            live_responder: Callable for upstream forwarding. Required when
                mode is record or auto.

        Returns:
            A new Broker instance with the loaded cassette.

        Raises:
            CassetteLoadError: When the store fails to load the cassette
                (e.g., missing file, corrupted data).
            LiveResponderRequiredError: When mode is record or auto and
                live_responder is not configured.
        """
        cassette = cassette_store.load()
        return cls(
            cassette=cassette,
            mode=mode,
            live_responder=live_responder,
            cassette_store=cassette_store,
        )

    @property
    def cassette(self) -> Cassette:
        """Get the cassette."""
        return self._cassette

    @property
    def mode(self) -> BrokerMode:
        """Get the broker mode."""
        return self._mode

    @property
    def live_responder(self) -> LiveResponder | None:
        """Get the live responder."""
        return self._live_responder

    @property
    def cassette_store(self) -> CassetteStore | None:
        """Get the cassette store."""
        return self._cassette_store

    def replay(self, request: InteractionRequest) -> Iterator[ResponseChunk]:
        """Replay recorded response for matching request.

        Args:
            request: The request to match and replay

        Yields:
            ResponseChunks in original recorded order

        Raises:
            InteractionNotFoundError: When no matching interaction exists
                and mode is replay.
        """
        # record mode: always forward to live, ignore cassette
        if self._mode == "record":
            yield from self._forward_and_record(request)
            return

        # replay/auto mode: try cassette first
        interaction = self.cassette.find_interaction(request.fingerprint())
        if interaction is not None:
            yield from interaction.response_chunks
            return

        # MISS handling
        if self._mode == "replay":
            raise InteractionNotFoundError(request)

        # auto mode MISS: forward to live
        yield from self._forward_and_record(request)

    def _forward_and_record(
        self, request: InteractionRequest
    ) -> Iterator[ResponseChunk]:
        """Forward request to live responder and record the interaction.

        Args:
            request: The request to forward

        Yields:
            ResponseChunks from live responder

        Raises:
            LiveResponderRequiredError: When live_responder is not configured.
        """
        if self._live_responder is None:
            raise LiveResponderRequiredError(self._mode)

        chunks = tuple(self._live_responder(request))
        self._record_interaction(request, chunks)
        if self._cassette_store is not None:
            self._cassette_store.save(self._cassette)
        yield from chunks

    def _record_interaction(
        self,
        request: InteractionRequest,
        response_chunks: tuple[ResponseChunk, ...],
    ) -> None:
        """Record a new interaction to the cassette.

        Creates a new Cassette with the interaction appended.

        Args:
            request: The request that was made
            response_chunks: The response chunks from live responder
        """
        interaction = Interaction(
            request=request,
            fingerprint=request.fingerprint(),
            response_chunks=response_chunks,
        )
        new_interactions = (*self._cassette.interactions, interaction)
        self._cassette = Cassette(interactions=new_interactions)

cassette property 

cassette: Cassette

Get the cassette.

mode property 

mode: BrokerMode

Get the broker mode.

live_responder property 

live_responder: LiveResponder | None

Get the live responder.

cassette_store property 

cassette_store: CassetteStore | None

Get the cassette store.

__init__ 

__init__(
    cassette: Cassette,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
    cassette_store: CassetteStore | None = None,
) -> None

Initialize broker with a cassette.

Parameters:

Name Type Description Default
cassette Cassette

The cassette containing recorded interactions

 required
mode BrokerMode

The broker mode (replay, record, or auto)

 'replay'
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto.

 None
cassette_store CassetteStore | None

Optional store for automatic cassette persistence

 None

Raises:

Type Description
LiveResponderRequiredError

When mode is record or auto and live_responder is not configured.
Source code in src/interposition/services.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    cassette: Cassette,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
    cassette_store: CassetteStore | None = None,
) -> None:
    """Initialize broker with a cassette.

    Args:
        cassette: The cassette containing recorded interactions
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when
            mode is record or auto.
        cassette_store: Optional store for automatic cassette persistence

    Raises:
        LiveResponderRequiredError: When mode is record or auto and
            live_responder is not configured.
    """
    if mode in ("record", "auto") and live_responder is None:
        raise LiveResponderRequiredError(mode)

    self._cassette = cassette
    self._mode = mode
    self._live_responder = live_responder
    self._cassette_store = cassette_store

from_store classmethod 

from_store(
    cassette_store: CassetteStore,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
) -> Self

Create a Broker by loading a cassette from a store.

Parameters:

Name Type Description Default
cassette_store CassetteStore

The store to load the cassette from

 required
mode BrokerMode

The broker mode (replay, record, or auto)

 'replay'
live_responder LiveResponder | None

Callable for upstream forwarding. Required when mode is record or auto.

 None

Returns:

Type Description
Self

A new Broker instance with the loaded cassette.

Raises:

Type Description
CassetteLoadError

When the store fails to load the cassette (e.g., missing file, corrupted data).
LiveResponderRequiredError

When mode is record or auto and live_responder is not configured.
Source code in src/interposition/services.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def from_store(
    cls,
    cassette_store: CassetteStore,
    mode: BrokerMode = "replay",
    live_responder: LiveResponder | None = None,
) -> Self:
    """Create a Broker by loading a cassette from a store.

    Args:
        cassette_store: The store to load the cassette from
        mode: The broker mode (replay, record, or auto)
        live_responder: Callable for upstream forwarding. Required when
            mode is record or auto.

    Returns:
        A new Broker instance with the loaded cassette.

    Raises:
        CassetteLoadError: When the store fails to load the cassette
            (e.g., missing file, corrupted data).
        LiveResponderRequiredError: When mode is record or auto and
            live_responder is not configured.
    """
    cassette = cassette_store.load()
    return cls(
        cassette=cassette,
        mode=mode,
        live_responder=live_responder,
        cassette_store=cassette_store,
    )

replay 

replay(
    request: InteractionRequest,
) -> Iterator[ResponseChunk]

Replay recorded response for matching request.

Parameters:

Name Type Description Default
request InteractionRequest

The request to match and replay

 required

Yields:

Type Description
ResponseChunk

ResponseChunks in original recorded order

Raises:

Type Description
InteractionNotFoundError

When no matching interaction exists and mode is replay.
Source code in src/interposition/services.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def replay(self, request: InteractionRequest) -> Iterator[ResponseChunk]:
    """Replay recorded response for matching request.

    Args:
        request: The request to match and replay

    Yields:
        ResponseChunks in original recorded order

    Raises:
        InteractionNotFoundError: When no matching interaction exists
            and mode is replay.
    """
    # record mode: always forward to live, ignore cassette
    if self._mode == "record":
        yield from self._forward_and_record(request)
        return

    # replay/auto mode: try cassette first
    interaction = self.cassette.find_interaction(request.fingerprint())
    if interaction is not None:
        yield from interaction.response_chunks
        return

    # MISS handling
    if self._mode == "replay":
        raise InteractionNotFoundError(request)

    # auto mode MISS: forward to live
    yield from self._forward_and_record(request)

CassetteStore

Bases: Protocol

Port for cassette persistence operations.

Implementations handle loading and saving cassettes to storage. The Broker calls save() automatically after recording new interactions.

Source code in src/interposition/services.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class CassetteStore(Protocol):
    """Port for cassette persistence operations.

    Implementations handle loading and saving cassettes to storage.
    The Broker calls save() automatically after recording new interactions.
    """

    def load(self) -> Cassette:
        """Load cassette from storage.

        Returns:
            The loaded Cassette instance.
        """
        ...

    def save(self, cassette: Cassette) -> None:
        """Save cassette to storage.

        Args:
            cassette: The cassette to persist.
        """
        ...

load 

load() -> Cassette

Load cassette from storage.

Returns:

Type Description
Cassette

The loaded Cassette instance.
Source code in src/interposition/services.py 
29
30
31
32
33
34
35
def load(self) -> Cassette:
    """Load cassette from storage.

    Returns:
        The loaded Cassette instance.
    """
    ...

save 

save(cassette: Cassette) -> None

Save cassette to storage.

Parameters:

Name Type Description Default
cassette Cassette

The cassette to persist.

 required
Source code in src/interposition/services.py 
37
38
39
40
41
42
43
def save(self, cassette: Cassette) -> None:
    """Save cassette to storage.

    Args:
        cassette: The cassette to persist.
    """
    ...

Stores

JsonFileCassetteStore

File-based cassette store using JSON format.

Attributes:

Name Type Description
path Path

Path to the JSON file for cassette storage.
Source code in src/interposition/stores.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class JsonFileCassetteStore:
    """File-based cassette store using JSON format.

    Attributes:
        path: Path to the JSON file for cassette storage.
    """

    def __init__(self, path: Path, *, create_if_missing: bool = False) -> None:
        """Initialize store with file path.

        Args:
            path: Path to the JSON file (will be created if doesn't exist).
            create_if_missing: If True, load() returns an empty Cassette
                when the file doesn't exist instead of raising CassetteLoadError.
        """
        self._path = path
        self._create_if_missing = create_if_missing

    @property
    def path(self) -> Path:
        """Get the file path."""
        return self._path

    def load(self) -> Cassette:
        """Load cassette from JSON file.

        Returns:
            Cassette instance loaded from file. If create_if_missing is True
            and the file doesn't exist, returns an empty Cassette.

        Raises:
            CassetteLoadError: If file doesn't exist (and create_if_missing
                is False), file is unreadable, or JSON is invalid.
        """
        try:
            json_str = self._path.read_text(encoding="utf-8")
        except FileNotFoundError as e:
            if self._create_if_missing:
                return Cassette(interactions=())
            raise CassetteLoadError(self._path, e) from e
        except OSError as e:
            raise CassetteLoadError(self._path, e) from e
        try:
            return Cassette.model_validate_json(json_str)
        except Exception as e:
            raise CassetteLoadError(self._path, e) from e

    def save(self, cassette: Cassette) -> None:
        """Save cassette to JSON file.

        Creates parent directories if they don't exist.

        Args:
            cassette: The cassette to persist.

        Raises:
            CassetteSaveError: If file write fails.
        """
        try:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            json_str = cassette.model_dump_json(indent=2)
            self._path.write_text(json_str, encoding="utf-8")
        except OSError as e:
            raise CassetteSaveError(self._path, e) from e

path property 

path: Path

Get the file path.

__init__ 

__init__(
    path: Path, *, create_if_missing: bool = False
) -> None

Initialize store with file path.

Parameters:

Name Type Description Default
path Path

Path to the JSON file (will be created if doesn't exist).

 required
create_if_missing bool

If True, load() returns an empty Cassette when the file doesn't exist instead of raising CassetteLoadError.

 False
Source code in src/interposition/stores.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, path: Path, *, create_if_missing: bool = False) -> None:
    """Initialize store with file path.

    Args:
        path: Path to the JSON file (will be created if doesn't exist).
        create_if_missing: If True, load() returns an empty Cassette
            when the file doesn't exist instead of raising CassetteLoadError.
    """
    self._path = path
    self._create_if_missing = create_if_missing

load 

load() -> Cassette

Load cassette from JSON file.

Returns:

Type Description
Cassette

Cassette instance loaded from file. If create_if_missing is True
Cassette

and the file doesn't exist, returns an empty Cassette.

Raises:

Type Description
CassetteLoadError

If file doesn't exist (and create_if_missing is False), file is unreadable, or JSON is invalid.
Source code in src/interposition/stores.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def load(self) -> Cassette:
    """Load cassette from JSON file.

    Returns:
        Cassette instance loaded from file. If create_if_missing is True
        and the file doesn't exist, returns an empty Cassette.

    Raises:
        CassetteLoadError: If file doesn't exist (and create_if_missing
            is False), file is unreadable, or JSON is invalid.
    """
    try:
        json_str = self._path.read_text(encoding="utf-8")
    except FileNotFoundError as e:
        if self._create_if_missing:
            return Cassette(interactions=())
        raise CassetteLoadError(self._path, e) from e
    except OSError as e:
        raise CassetteLoadError(self._path, e) from e
    try:
        return Cassette.model_validate_json(json_str)
    except Exception as e:
        raise CassetteLoadError(self._path, e) from e

save 

save(cassette: Cassette) -> None

Save cassette to JSON file.

Creates parent directories if they don't exist.

Parameters:

Name Type Description Default
cassette Cassette

The cassette to persist.

 required

Raises:

Type Description
CassetteSaveError

If file write fails.
Source code in src/interposition/stores.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def save(self, cassette: Cassette) -> None:
    """Save cassette to JSON file.

    Creates parent directories if they don't exist.

    Args:
        cassette: The cassette to persist.

    Raises:
        CassetteSaveError: If file write fails.
    """
    try:
        self._path.parent.mkdir(parents=True, exist_ok=True)
        json_str = cassette.model_dump_json(indent=2)
        self._path.write_text(json_str, encoding="utf-8")
    except OSError as e:
        raise CassetteSaveError(self._path, e) from e

Models

Cassette

Bases: BaseModel

In-memory collection of recorded interactions.

Attributes:

Name Type Description
interactions tuple[Interaction, ...]

Ordered sequence of interactions
Source code in src/interposition/models.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class Cassette(BaseModel):
    """In-memory collection of recorded interactions.

    Attributes:
        interactions: Ordered sequence of interactions
    """

    model_config = ConfigDict(frozen=True)

    interactions: tuple[Interaction, ...]
    _index: dict[RequestFingerprint, int] = PrivateAttr(default_factory=dict)

    @model_validator(mode="after")
    def build_index(self) -> Self:
        """Build fingerprint index for efficient lookup."""
        index: dict[RequestFingerprint, int] = {}
        for i, interaction in enumerate(self.interactions):
            # Only store first occurrence of each fingerprint
            if interaction.fingerprint not in index:
                index[interaction.fingerprint] = i
        # Use object.__setattr__ to modify frozen model
        object.__setattr__(self, "_index", index)
        return self

    def find_interaction(self, fingerprint: RequestFingerprint) -> Interaction | None:
        """Find first interaction matching fingerprint.

        Args:
            fingerprint: Request fingerprint to search for

        Returns:
            Matching Interaction or None if not found
        """
        position = self._index.get(fingerprint)
        if position is None:
            return None
        return self.interactions[position]

build_index 

build_index() -> Self

Build fingerprint index for efficient lookup.

Source code in src/interposition/models.py 
203
204
205
206
207
208
209
210
211
212
213
@model_validator(mode="after")
def build_index(self) -> Self:
    """Build fingerprint index for efficient lookup."""
    index: dict[RequestFingerprint, int] = {}
    for i, interaction in enumerate(self.interactions):
        # Only store first occurrence of each fingerprint
        if interaction.fingerprint not in index:
            index[interaction.fingerprint] = i
    # Use object.__setattr__ to modify frozen model
    object.__setattr__(self, "_index", index)
    return self

find_interaction 

find_interaction(
    fingerprint: RequestFingerprint,
) -> Interaction | None

Find first interaction matching fingerprint.

Parameters:

Name Type Description Default
fingerprint RequestFingerprint

Request fingerprint to search for

 required

Returns:

Type Description
Interaction | None

Matching Interaction or None if not found
Source code in src/interposition/models.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
def find_interaction(self, fingerprint: RequestFingerprint) -> Interaction | None:
    """Find first interaction matching fingerprint.

    Args:
        fingerprint: Request fingerprint to search for

    Returns:
        Matching Interaction or None if not found
    """
    position = self._index.get(fingerprint)
    if position is None:
        return None
    return self.interactions[position]

Interaction

Bases: BaseModel

Complete request-response pair for replay.

Attributes:

Name Type Description
request InteractionRequest

The original InteractionRequest
fingerprint RequestFingerprint

Precomputed request fingerprint for matching
response_chunks tuple[ResponseChunk, ...]

Ordered sequence of response chunks
metadata tuple[tuple[str, str], ...]

Optional interaction metadata as (key, value) pairs. Examples: recording timestamp, session ID, test scenario name. Useful for debugging and tracing recorded interactions. Default is empty tuple.
Source code in src/interposition/models.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class Interaction(BaseModel):
    """Complete request-response pair for replay.

    Attributes:
        request: The original InteractionRequest
        fingerprint: Precomputed request fingerprint for matching
        response_chunks: Ordered sequence of response chunks
        metadata: Optional interaction metadata as (key, value) pairs.
            Examples: recording timestamp, session ID, test scenario name.
            Useful for debugging and tracing recorded interactions.
            Default is empty tuple.
    """

    model_config = ConfigDict(frozen=True)

    request: InteractionRequest
    fingerprint: RequestFingerprint
    response_chunks: tuple[ResponseChunk, ...]
    metadata: tuple[tuple[str, str], ...] = ()

    @model_validator(mode="after")
    def validate_interaction(self) -> Self:
        """Validate interaction integrity.

        Raises:
            InteractionValidationError: If fingerprint doesn't match request
                or chunks aren't sequential
        """
        # Verify fingerprint matches request
        expected_fingerprint = self.request.fingerprint()
        if self.fingerprint != expected_fingerprint:
            msg = (
                f"Fingerprint does not match request: "
                f"expected {expected_fingerprint.value}, got {self.fingerprint.value}"
            )
            raise InteractionValidationError(msg)

        # Verify response chunks are sequentially ordered
        if not self.response_chunks:
            msg = "Response chunks cannot be empty"
            raise InteractionValidationError(msg)

        if self.response_chunks[0].sequence != 0:
            msg = "Response chunks must start at sequence 0"
            raise InteractionValidationError(msg)

        for i, chunk in enumerate(self.response_chunks):
            if chunk.sequence != i:
                msg = "Response chunks must be sequential with no gaps"
                raise InteractionValidationError(msg)

        return self

validate_interaction 

validate_interaction() -> Self

Validate interaction integrity.

Raises:

Type Description
InteractionValidationError

If fingerprint doesn't match request or chunks aren't sequential
Source code in src/interposition/models.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@model_validator(mode="after")
def validate_interaction(self) -> Self:
    """Validate interaction integrity.

    Raises:
        InteractionValidationError: If fingerprint doesn't match request
            or chunks aren't sequential
    """
    # Verify fingerprint matches request
    expected_fingerprint = self.request.fingerprint()
    if self.fingerprint != expected_fingerprint:
        msg = (
            f"Fingerprint does not match request: "
            f"expected {expected_fingerprint.value}, got {self.fingerprint.value}"
        )
        raise InteractionValidationError(msg)

    # Verify response chunks are sequentially ordered
    if not self.response_chunks:
        msg = "Response chunks cannot be empty"
        raise InteractionValidationError(msg)

    if self.response_chunks[0].sequence != 0:
        msg = "Response chunks must start at sequence 0"
        raise InteractionValidationError(msg)

    for i, chunk in enumerate(self.response_chunks):
        if chunk.sequence != i:
            msg = "Response chunks must be sequential with no gaps"
            raise InteractionValidationError(msg)

    return self

InteractionRequest

Bases: BaseModel

Structured representation of a protocol-agnostic request.

Attributes:

Name Type Description
protocol str

Protocol identifier (e.g., "grpc", "graphql", "mqtt")
action str

Action/method name (e.g., "ListUsers", "query", "publish")
target str

Target resource (e.g., "users.UserService", "topic/sensors")
headers tuple[tuple[str, str], ...]

Request headers as immutable sequence of key-value pairs
body bytes

Request body content as bytes
Source code in src/interposition/models.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class InteractionRequest(BaseModel):
    """Structured representation of a protocol-agnostic request.

    Attributes:
        protocol: Protocol identifier (e.g., "grpc", "graphql", "mqtt")
        action: Action/method name (e.g., "ListUsers", "query", "publish")
        target: Target resource (e.g., "users.UserService", "topic/sensors")
        headers: Request headers as immutable sequence of key-value pairs
        body: Request body content as bytes
    """

    model_config = ConfigDict(frozen=True)

    protocol: str
    action: str
    target: str
    headers: tuple[tuple[str, str], ...]
    body: bytes

    def fingerprint(self) -> RequestFingerprint:
        """Generate stable fingerprint for efficient matching.

        Returns:
            RequestFingerprint derived from all request fields.
        """
        return RequestFingerprint.from_request(self)

fingerprint 

fingerprint() -> RequestFingerprint

Generate stable fingerprint for efficient matching.

Returns:

Type Description
RequestFingerprint

RequestFingerprint derived from all request fields.
Source code in src/interposition/models.py 
67
68
69
70
71
72
73
def fingerprint(self) -> RequestFingerprint:
    """Generate stable fingerprint for efficient matching.

    Returns:
        RequestFingerprint derived from all request fields.
    """
    return RequestFingerprint.from_request(self)

ResponseChunk

Bases: BaseModel

Discrete piece of response data.

Attributes:

Name Type Description
data bytes

Chunk payload as bytes
sequence int

Zero-based chunk position in response stream
metadata tuple[tuple[str, str], ...]

Optional chunk metadata as (key, value) string pairs. Examples: timing info, encoding, content-type for this chunk. Default is empty tuple.
Source code in src/interposition/models.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ResponseChunk(BaseModel):
    """Discrete piece of response data.

    Attributes:
        data: Chunk payload as bytes
        sequence: Zero-based chunk position in response stream
        metadata: Optional chunk metadata as (key, value) string pairs.
            Examples: timing info, encoding, content-type for this chunk.
            Default is empty tuple.
    """

    model_config = ConfigDict(frozen=True)

    data: bytes
    sequence: int
    metadata: tuple[tuple[str, str], ...] = ()

RequestFingerprint

Bases: BaseModel

Stable unique identifier for request matching.

Attributes:

Name Type Description
value str

SHA-256 hash of canonicalized request fields
Source code in src/interposition/models.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class RequestFingerprint(BaseModel):
    """Stable unique identifier for request matching.

    Attributes:
        value: SHA-256 hash of canonicalized request fields
    """

    model_config = ConfigDict(frozen=True)

    value: str

    @field_validator("value")
    @classmethod
    def validate_sha256_hex(cls, v: str) -> str:
        """Validate that value is a valid SHA-256 hex string.

        Args:
            v: The fingerprint value to validate

        Returns:
            The validated value

        Raises:
            ValueError: If value is not exactly 64 hex characters
        """
        if len(v) != SHA256_HEX_LENGTH:
            msg = f"SHA-256 hex must be exactly {SHA256_HEX_LENGTH} characters"
            raise ValueError(msg)
        if not all(c in "0123456789abcdef" for c in v):
            msg = "Invalid hex characters in fingerprint"
            raise ValueError(msg)
        return v

    @classmethod
    def from_request(cls, request: InteractionRequest) -> Self:
        """Create fingerprint from InteractionRequest.

        Args:
            request: The request to fingerprint

        Returns:
            RequestFingerprint with SHA-256 hash value
        """
        # Canonical order: protocol, action, target, headers, body
        # Preserve header ordering to avoid normalization.
        canonical_data = [
            request.protocol,
            request.action,
            request.target,
            request.headers,
            request.body.hex(),
        ]
        canonical = json.dumps(
            canonical_data,
            separators=_CANONICAL_JSON_SEPARATORS,
            sort_keys=_CANONICAL_JSON_SORT_KEYS,
        )
        hash_value = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        return cls(value=hash_value)

validate_sha256_hex classmethod 

validate_sha256_hex(v: str) -> str

Validate that value is a valid SHA-256 hex string.

Parameters:

Name Type Description Default
v str

The fingerprint value to validate

 required

Returns:

Type Description
str

The validated value

Raises:

Type Description
ValueError

If value is not exactly 64 hex characters
Source code in src/interposition/models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@field_validator("value")
@classmethod
def validate_sha256_hex(cls, v: str) -> str:
    """Validate that value is a valid SHA-256 hex string.

    Args:
        v: The fingerprint value to validate

    Returns:
        The validated value

    Raises:
        ValueError: If value is not exactly 64 hex characters
    """
    if len(v) != SHA256_HEX_LENGTH:
        msg = f"SHA-256 hex must be exactly {SHA256_HEX_LENGTH} characters"
        raise ValueError(msg)
    if not all(c in "0123456789abcdef" for c in v):
        msg = "Invalid hex characters in fingerprint"
        raise ValueError(msg)
    return v

from_request classmethod 

from_request(request: InteractionRequest) -> Self

Create fingerprint from InteractionRequest.

Parameters:

Name Type Description Default
request InteractionRequest

The request to fingerprint

 required

Returns:

Type Description
Self

RequestFingerprint with SHA-256 hash value
Source code in src/interposition/models.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def from_request(cls, request: InteractionRequest) -> Self:
    """Create fingerprint from InteractionRequest.

    Args:
        request: The request to fingerprint

    Returns:
        RequestFingerprint with SHA-256 hash value
    """
    # Canonical order: protocol, action, target, headers, body
    # Preserve header ordering to avoid normalization.
    canonical_data = [
        request.protocol,
        request.action,
        request.target,
        request.headers,
        request.body.hex(),
    ]
    canonical = json.dumps(
        canonical_data,
        separators=_CANONICAL_JSON_SEPARATORS,
        sort_keys=_CANONICAL_JSON_SORT_KEYS,
    )
    hash_value = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return cls(value=hash_value)

Exceptions

InterpositionError

Bases: Exception

Base class for all interposition exceptions.

Source code in src/interposition/errors.py 
13
14
class InterpositionError(Exception):
    """Base class for all interposition exceptions."""

CassetteSaveError

Bases: InterpositionError

Raised when cassette persistence fails.

Source code in src/interposition/errors.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
class CassetteSaveError(InterpositionError):
    """Raised when cassette persistence fails."""

    def __init__(self, path: Path, cause: Exception) -> None:
        """Initialize with the path and underlying cause.

        Args:
            path: The file path where save failed
            cause: The underlying exception that caused the failure
        """
        super().__init__(f"Failed to save cassette to {path}: {cause}")
        self.path: Path = path
        self.__cause__ = cause

__init__ 

__init__(path: Path, cause: Exception) -> None

Initialize with the path and underlying cause.

Parameters:

Name Type Description Default
path Path

The file path where save failed

 required
cause Exception

The underlying exception that caused the failure

 required
Source code in src/interposition/errors.py 
64
65
66
67
68
69
70
71
72
73
def __init__(self, path: Path, cause: Exception) -> None:
    """Initialize with the path and underlying cause.

    Args:
        path: The file path where save failed
        cause: The underlying exception that caused the failure
    """
    super().__init__(f"Failed to save cassette to {path}: {cause}")
    self.path: Path = path
    self.__cause__ = cause

InteractionNotFoundError

Bases: InterpositionError

Raised when no matching interaction is found in cassette.

Source code in src/interposition/errors.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class InteractionNotFoundError(InterpositionError):
    """Raised when no matching interaction is found in cassette."""

    def __init__(self, request: InteractionRequest) -> None:
        """Initialize with request that failed to match.

        Args:
            request: The unmatched request
        """
        super().__init__(
            f"No matching interaction for {request.protocol}:"
            f"{request.action}:{request.target}"
        )
        self.request: InteractionRequest = request

__init__ 

__init__(request: InteractionRequest) -> None

Initialize with request that failed to match.

Parameters:

Name Type Description Default
request InteractionRequest

The unmatched request

 required
Source code in src/interposition/errors.py 
20
21
22
23
24
25
26
27
28
29
30
def __init__(self, request: InteractionRequest) -> None:
    """Initialize with request that failed to match.

    Args:
        request: The unmatched request
    """
    super().__init__(
        f"No matching interaction for {request.protocol}:"
        f"{request.action}:{request.target}"
    )
    self.request: InteractionRequest = request

InteractionValidationError

Bases: InterpositionError, ValueError

Raised when interaction validation fails.

Source code in src/interposition/models.py 
26
27
class InteractionValidationError(InterpositionError, ValueError):
    """Raised when interaction validation fails."""

LiveResponderRequiredError

Bases: InterpositionError

Raised when live_responder is required but not configured.

Source code in src/interposition/errors.py 
33
34
35
36
37
38
39
40
41
42
43
class LiveResponderRequiredError(InterpositionError):
    """Raised when live_responder is required but not configured."""

    def __init__(self, mode: str) -> None:
        """Initialize with the mode that requires live_responder.

        Args:
            mode: The broker mode that requires live_responder
        """
        super().__init__(f"live_responder is required for {mode} mode")
        self.mode: str = mode

__init__ 

__init__(mode: str) -> None

Initialize with the mode that requires live_responder.

Parameters:

Name Type Description Default
mode str

The broker mode that requires live_responder

 required
Source code in src/interposition/errors.py 
36
37
38
39
40
41
42
43
def __init__(self, mode: str) -> None:
    """Initialize with the mode that requires live_responder.

    Args:
        mode: The broker mode that requires live_responder
    """
    super().__init__(f"live_responder is required for {mode} mode")
    self.mode: str = mode