### Setup and Test Interposition Project in Bash

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Contains bash commands for cloning the interposition repository, installing development dependencies using `uv`, and running tests with `nox`. This is a guide for developers setting up the project locally.

```bash
# Clone and install
git clone https://github.com/osoekawaitlab/interposition.git
cd interposition
uv pip install -e . --group=dev

# Run tests
nox -s tests
```

--------------------------------

### Development Setup

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Details the prerequisites and steps required for setting up the development environment, including installing dependencies and running tests.

```APIDOC
## Development

### Prerequisites

- Python 3.10+
- [uv](https://github.com/astral-sh/uv) (recommended)

### Setup & Testing

```bash
# Clone and install
git clone https://github.com/osoekawaitlab/interposition.git
cd interposition
uv pip install -e . --group=dev

# Run tests
nox -s tests
```
```

--------------------------------

### Install Interposition using pip

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This command installs the interposition library using pip, the Python package installer. Ensure you have pip installed and configured correctly.

```bash
pip install interposition
```

--------------------------------

### Create Broker with CassetteStore using from_store

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Shows a convenient way to initialize a Broker with a CassetteStore using the `Broker.from_store()` class method. This method handles loading the cassette and creating the broker instance in a single step, simplifying setup for persistence workflows.

```python
store = JsonFileCassetteStore(
    Path("cassettes/my_test.json"),
    create_if_missing=True,
)
broker = Broker.from_store(store, mode="auto", live_responder=my_live_responder)
```

--------------------------------

### Record Interactions with Broker

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Shows how to use the Broker in 'record' mode to capture new interactions. It starts with an empty cassette, forwards all requests to the live responder, records them, and then demonstrates saving the updated cassette to a JSON file.

```python
# Start with empty cassette
cassette = Cassette(interactions=())

broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=my_live_responder,
)

# All requests are forwarded and recorded
response = list(broker.replay(request))

# Save the updated cassette
with open("cassette.json", "w") as f:
    f.write(broker.cassette.model_dump_json(indent=2))
```

--------------------------------

### Implement Live Responder Callable

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Provides an example of a live responder function. This callable is essential for 'record' and 'auto' modes, as it forwards requests to the actual backend and yields response chunks. It requires an external HTTP client implementation.

```python
from interposition import (
    Broker,
    Cassette,
    InteractionRequest,
    ResponseChunk,
)
from collections.abc import Iterable

def my_live_responder(request: InteractionRequest) -> Iterable[ResponseChunk]:
    """Forward request to actual backend and yield response chunks."""
    # Your actual implementation here
    response = your_http_client.request(
        method=request.action,
        url=request.target,
        headers=dict(request.headers),
        data=request.body,
    )
    yield ResponseChunk(data=response.content, sequence=0)
```

--------------------------------

### Error Handling

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This section details the custom error types provided by the interposition library, all inheriting from `InterpositionError`. It includes examples of how to catch specific errors like `InteractionNotFoundError`, `LiveResponderRequiredError`, `InteractionValidationError`, `CassetteLoadError`, and `CassetteSaveError`.

```APIDOC
## Error Handling

All interposition exceptions inherit from `InterpositionError`, allowing you to catch all domain errors with a single handler:

```python
from interposition import InterpositionError

try:
    broker.replay(request)
except InterpositionError as e:
    print(f"Interposition error: {e}")
```

**InteractionNotFoundError**: Raised when no matching interaction exists in `replay` mode:

```python
from interposition import InteractionNotFoundError

try:
    broker.replay(unknown_request)
except InteractionNotFoundError as e:
    print(f"Not recorded: {e.request.target}")
```

**LiveResponderRequiredError**: Raised at broker construction when `record` or `auto` mode is used without a `live_responder`:

```python
from interposition import LiveResponderRequiredError

try:
    Broker(cassette=cassette, mode="auto")  # No live_responder!
except LiveResponderRequiredError as e:
    print(f"live_responder required for {e.mode} mode")
```

**InteractionValidationError**: Raised when an `Interaction` fails validation (e.g., fingerprint mismatch or invalid response chunk sequence):

```python
from interposition import Interaction, InteractionValidationError

try:
    # This will fail: fingerprint doesn't match request
    interaction = Interaction(
        request=request,
        fingerprint=wrong_fingerprint,  # Mismatch!
        response_chunks=chunks,
    )
except InteractionValidationError as e:
    print(f"Validation failed: {e}")
```

**CassetteLoadError**: Raised when `JsonFileCassetteStore.load()` fails (file not found, permission denied, corrupted JSON, etc.):

```python
from pathlib import Path
from interposition import CassetteLoadError, JsonFileCassetteStore

store = JsonFileCassetteStore(Path("cassettes/missing.json"))

try:
    cassette = store.load()
except CassetteLoadError as e:
    print(f"Failed to load from {e.path}: {e.__cause__}")
```

**CassetteSaveError**: Raised when `JsonFileCassetteStore.save()` fails due to I/O errors (permission denied, disk full, etc.):

```python
from pathlib import Path
from interposition import CassetteSaveError, JsonFileCassetteStore

store = JsonFileCassetteStore(Path("/readonly/cassette.json"))

try:
    store.save(cassette)
except CassetteSaveError as e:
    print(f"Failed to save to {e.path}: {e.__cause__}")
```
```

--------------------------------

### Basic Replay Mode with Interposition Broker

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Illustrates the basic replay functionality of the `Broker` class. The default mode is 'replay', which means it will only return recorded responses from the provided cassette. It shows how to set up a cassette and a broker for replay.

```python
from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionRequest,
    ResponseChunk,
    InteractionNotFoundError,
)

# Setup cassette with recorded interactions
request = InteractionRequest(
    protocol="http",
    action="GET",
    target="https://api.example.com/users/42",
    headers=(),
    body=b"",
)

interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=(ResponseChunk(data=b'{"id": 42, "name": "Alice"}', sequence=0),),
)

cassette = Cassette(interactions=(interaction,))
broker = Broker(cassette=cassette)  # mode="replay" is default
```

--------------------------------

### Access Interposition Package Version in Python

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Provides a Python code snippet to programmatically access the installed `interposition` package version using `__version__`. This allows for conditional logic based on the library's version.

```python
from interposition import __version__

if __version__ < "0.2.0":
    print("Auto mode is not supported")
else:
    print("Auto mode is supported")
```

--------------------------------

### Create Protocol-Agnostic Request with Interposition

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates the creation of `InteractionRequest` objects for various protocols like HTTP, SQL, and MQTT. It shows how to define the protocol, action, target, headers, and body, and how to generate a fingerprint for matching.

```python
from interposition import InteractionRequest

# HTTP request example
http_request = InteractionRequest(
    protocol="http",
    action="GET",
    target="https://api.example.com/users/42",
    headers=(("Authorization", "Bearer token123"),),
    body=b"",
)

# SQL database query example
sql_request = InteractionRequest(
    protocol="postgres",
    action="SELECT",
    target="users_table",
    headers=(),
    body=b"SELECT id, name FROM users WHERE id = 42",
)

# MQTT/PubSub message example
mqtt_request = InteractionRequest(
    protocol="mqtt",
    action="subscribe",
    target="sensors/temp/room1",
    headers=(("qos", "1"),),
    body=b"",
)

# Generate fingerprint for matching
fingerprint = http_request.fingerprint()
print(fingerprint.value)  # SHA-256 hash string
```

--------------------------------

### Basic Interposition Request and Replay

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates the core functionality of Interposition by creating an InteractionRequest, packaging it into an Interaction and Cassette, and then using a Broker and Opener to replay the interaction.

```python
from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionRequest,
    ResponseChunk,
    build_opener
)

request = InteractionRequest(
    protocol="http",
    action="GET",
    target="https://api.example.com/users/1",
    headers=(),
    body=b"",
)

interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=(ResponseChunk(data=b'{"id": 1, "name": "Alice"}', sequence=0),),
)

cassette = Cassette(interactions=(interaction,))
broker = Broker(cassette)

handler = InterpositionHandler(broker)
opener = build_opener(handler)

response = opener.open("https://api.example.com/users/1")
print(response.read())  # b'{"id": 1, "name": "Alice"}'
```

--------------------------------

### Replay Interactions and Handle Missing Ones - Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates replaying recorded interactions and handling cases where an interaction is not found. It shows how to access response data and catch specific exceptions.

```python
from interposition import Broker, InteractionRequest, InteractionNotFoundError

# Assuming 'broker' is an initialized Broker instance
# response_chunks = list(broker.replay(request))
# print(response_chunks[0].data)  # b'{"id": 42, "name": "Alice"}'

# Handle missing interactions
unknown_request = InteractionRequest(
    protocol="http", action="GET", target="/unknown", headers=(), body=b""
)
try:
    list(broker.replay(unknown_request))
except InteractionNotFoundError as e:
    print(f"Not recorded: {e.request.target}")  # Not recorded: /unknown
```

--------------------------------

### Pytest Fixture for Interposition Broker in Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Provides a reusable pytest fixture to load an Interposition cassette from a JSON file and create a Broker instance for use in tests. It demonstrates how to use this fixture to mock network requests within a test function.

```python
import pytest
from interposition import Broker, Cassette, InteractionRequest, ResponseChunk

@pytest.fixture
def cassette_broker():
    """Load cassette from JSON file for test replay."""
    with open("tests/fixtures/my_cassette.json", "rb") as f:
        cassette = Cassette.model_validate_json(f.read())
    return Broker(cassette)

def test_user_service(cassette_broker, monkeypatch):
    # Create adapter that delegates to Interposition
    def mock_fetch(url: str) -> bytes:
        request = InteractionRequest(
            protocol="http",
            action="GET",
            target=url,
            headers=(),
            body=b"",
        )
        chunks = list(cassette_broker.replay(request))
        return chunks[0].data

    # Inject the adapter
    monkeypatch.setattr("my_app.client.fetch", mock_fetch)

    # Run your test
    from my_app import get_user_name
    assert get_user_name(42) == "Alice"

```

--------------------------------

### Auto Mode: Hybrid Replay and Record - Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates the 'auto' mode for hybrid workflows. It replays interactions if available in the cassette, otherwise it forwards the request to a live responder and records the response.

```python
from interposition import Broker, Cassette, InteractionRequest, ResponseChunk
from collections.abc import Iterable

def live_responder(request: InteractionRequest) -> Iterable[ResponseChunk]:
    """Your actual HTTP client implementation"""
    yield ResponseChunk(data=b'{"live": true}', sequence=0)

# Load existing cassette (may have some interactions)
# with open("cassette.json") as f:
#     cassette = Cassette.model_validate_json(f.read())

# Placeholder for cassette loading
cassette = Cassette(interactions=())

broker = Broker(
    cassette=cassette,
    mode="auto",
    live_responder=live_responder,
)

# Returns recorded response if exists, otherwise forwards and records
request = InteractionRequest(
    protocol="http",
    action="GET",
    target="https://api.example.com/data",
    headers=(),
    body=b"",
)
response = list(broker.replay(request))

# Save updated cassette after recording new interactions
# with open("cassette.json", "w") as f:
#     f.write(broker.cassette.model_dump_json(indent=2))
```

--------------------------------

### Record Mode: Capture New Interactions - Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Illustrates how to use the 'record' mode to capture new network interactions. It defines a live responder to forward requests to a backend and records the responses in a cassette.

```python
from collections.abc import Iterable
from interposition import (
    Broker,
    Cassette,
    InteractionRequest,
    ResponseChunk,
)
import urllib.request

# Define live responder that forwards to actual backend
def my_live_responder(request: InteractionRequest) -> Iterable[ResponseChunk]:
    """Forward request to actual backend and yield response chunks."""
    req = urllib.request.Request(
        url=request.target,
        method=request.action,
        data=request.body if request.body else None,
    )
    with urllib.request.urlopen(req) as response:
        yield ResponseChunk(data=response.read(), sequence=0)

# Start with empty cassette
cassette = Cassette(interactions=())

broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=my_live_responder,
)

# All requests are forwarded and recorded
request = InteractionRequest(
    protocol="http",
    action="GET",
    target="https://httpbin.org/get",
    headers=(),
    body=b"",
)
response = list(broker.replay(request))

# Access updated cassette with recorded interactions
print(len(broker.cassette.interactions))  # 1
```

--------------------------------

### Utilize JsonFileCassetteStore for JSON Persistence

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Demonstrates the use of the built-in JsonFileCassetteStore for automatic cassette persistence to a JSON file. It shows how to initialize the store, load cassettes, and configure the Broker to automatically save cassettes after recording.

```python
from pathlib import Path
from interposition import Broker, Cassette, JsonFileCassetteStore

# Create store pointing to a JSON file
store = JsonFileCassetteStore(Path("cassettes/my_test.json"))

# Load existing cassette (raises CassetteLoadError if not exists)
cassette = store.load()

# Or start with empty cassette
cassette = Cassette(interactions=())

# Create broker with automatic persistence
broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=my_live_responder,
    cassette_store=store,  # Auto-saves after each recording
)

# After replay, cassette is automatically saved to file
response = list(broker.replay(request))
```

--------------------------------

### Use Broker in Auto Mode for Hybrid Workflows

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Illustrates the 'auto' mode of the Broker, which combines replaying recorded responses with live recording. It loads an existing cassette, and if a response is not found, it forwards the request to the live responder and records the interaction.

```python
# Load existing cassette (may be empty or partial)
with open("cassette.json") as f:
    cassette = Cassette.model_validate_json(f.read())

broker = Broker(
    cassette=cassette,
    mode="auto",
    live_responder=my_live_responder,
)

# Returns recorded response if exists, otherwise forwards and records
response = list(broker.replay(request))
```

--------------------------------

### Manually Construct Interposition Interaction and Cassette

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This Python code illustrates how to manually construct an `Interaction` and `Cassette` for Interposition. It involves defining the `InteractionRequest`, `ResponseChunk`, and then combining them into a `Cassette` object for replay.

```python
from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionRequest,
    ResponseChunk,
)

# 1. Define the Request
request = InteractionRequest(
    protocol="api",
    action="query",
    target="users/42",
    headers=(),
    body=b"",
)

# 2. Define the Response
chunks = (
    ResponseChunk(data=b'{"id": 42, "name": "Alice"}', sequence=0),
)

# 3. Create Interaction & Cassette
interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=chunks,
)
cassette = Cassette(interactions=(interaction,))

# 4. Replay
broker = Broker(cassette=cassette)
response = list(broker.replay(request))
```

--------------------------------

### Define Broker Mode Type Hint

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Demonstrates how to use the BrokerMode type alias for type hinting when setting the broker's operating mode. This improves code readability and maintainability.

```python
from interposition import BrokerMode

mode: BrokerMode = "auto"
```

--------------------------------

### HTTP Proxy Server with Interposition

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Implements an HTTP proxy server using Python's http.server and socketserver modules, integrating with Interposition to handle requests. This allows any HTTP client to be proxied through Interposition.

```python
import http.server
import socketserver
from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionNotFoundError,
    InteractionRequest,
    ResponseChunk,
)

class InterpositionProxyHandler(http.server.BaseHTTPRequestHandler):
    """HTTP proxy handler that delegates to Interposition."""

    broker: Broker  # Set by server setup

    def do_GET(self) -> None:
        self.handle_request("GET")

    def do_POST(self) -> None:
        self.handle_request("POST")

    def handle_request(self, method: str) -> None:
        # Capture request body
        length = int(self.headers.get("content-length", 0))
        body = self.rfile.read(length) if length > 0 else b""

        inter_req = InteractionRequest(
            protocol="http",
            action=method,
            target=self.path,
            headers=(),
            body=body,
        )

        try:
            response_gen = self.broker.replay(inter_req)
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            for chunk in response_gen:
                self.wfile.write(chunk.data)
        except InteractionNotFoundError:
            self.send_error(404, "Interaction not recorded")

# Setup and run proxy server
request = InteractionRequest(
    protocol="http",
    action="GET",
    target="http://example.com/api/data",
    headers=(),
    body=b"",
)

interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=(ResponseChunk(data=b'{"status": "ok"}', sequence=0),),
)

cassette = Cassette(interactions=(interaction,))
InterpositionProxyHandler.broker = Broker(cassette)

# Start server: curl -x http://localhost:8080 http://example.com/api/data
with socketserver.TCPServer(("", 8080), InterpositionProxyHandler) as httpd:
    httpd.serve_forever()
```

--------------------------------

### Configure JsonFileCassetteStore to Create Missing Files

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Explains how to configure JsonFileCassetteStore with `create_if_missing=True`. This option ensures that an empty Cassette is returned when loading from a non-existent file, which is beneficial for record and auto modes where the file will be created on the first save.

```python
store = JsonFileCassetteStore(
    Path("cassettes/my_test.json"),
    create_if_missing=True,
)
cassette = store.load()  # Returns empty Cassette if file doesn't exist
```

--------------------------------

### JsonFileCassetteStore: File-Based Cassette Persistence - Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Shows how to use JsonFileCassetteStore for persistent cassette storage. It handles automatic JSON serialization, deserialization, and creation of parent directories if they don't exist.

```python
from pathlib import Path
from interposition import (
    Broker,
    Cassette,
    JsonFileCassetteStore,
    InteractionRequest,
    ResponseChunk,
    CassetteLoadError,
)
from collections.abc import Iterable

def live_responder(request: InteractionRequest) -> Iterable[ResponseChunk]:
    yield ResponseChunk(data=b'{"response": "data"}', sequence=0)

# Create store with automatic empty cassette creation
store = JsonFileCassetteStore(
    Path("cassettes/my_test.json"),
    create_if_missing=True,  # Returns empty Cassette if file doesn't exist
)

# Load cassette (creates empty if missing)
cassette = store.load()

# Create broker with automatic persistence
broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=live_responder,
    cassette_store=store,  # Auto-saves after each recording
)

# After replay, cassette is automatically saved to file
request = InteractionRequest(
    protocol="http", action="GET", target="/api/test", headers=(), body=b""
)
response = list(broker.replay(request))

# Alternative: Create broker directly from store
broker_from_store = Broker.from_store(
    store,
    mode="auto",
    live_responder=live_responder,
)
```

--------------------------------

### Create Request-Response Pairs with Interposition

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Shows how to create an `Interaction` object, which represents a complete request-response pair. This includes associating an `InteractionRequest` with its fingerprint and `ResponseChunk`s, along with optional metadata.

```python
from interposition import Interaction, InteractionRequest, ResponseChunk

request = InteractionRequest(
    protocol="api",
    action="query",
    target="users/42",
    headers=(),
    body=b"",
)

chunks = (
    ResponseChunk(data=b'{"id": 42, "name": "Alice"}', sequence=0),
)

# Create interaction with automatic fingerprint validation
interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=chunks,
    metadata=(("recorded_at", "2024-01-15T10:30:00Z"),)
)
```

--------------------------------

### Manage Recorded Interactions with Interposition Cassette

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates the creation and usage of a `Cassette`, which is an in-memory collection of recorded interactions. It highlights the O(1) fingerprint-based lookup for finding interactions within the cassette.

```python
from interposition import Cassette, Interaction, InteractionRequest, ResponseChunk

# Create multiple interactions
request1 = InteractionRequest( protocol="http", action="GET", target="/users/1", headers=(), body=b"" )
request2 = InteractionRequest( protocol="http", action="GET", target="/users/2", headers=(), body=b"" )

interaction1 = Interaction(
    request=request1,
    fingerprint=request1.fingerprint(),
    response_chunks=(ResponseChunk(data=b'{"name": "Alice"}', sequence=0),),
)
interaction2 = Interaction(
    request=request2,
    fingerprint=request2.fingerprint(),
    response_chunks=(ResponseChunk(data=b'{"name": "Bob"}', sequence=0),),
)

# Create cassette with interactions
cassette = Cassette(interactions=(interaction1, interaction2))

# Find interaction by fingerprint (O(1) lookup)
found = cassette.find_interaction(request1.fingerprint())
if found:
    print(found.response_chunks[0].data)  # b'{"name": "Alice"}'
```

--------------------------------

### urllib HTTP Handler Adapter using Interposition in Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Implements a custom urllib.request handler that integrates with the Interposition Broker. This allows standard Python HTTP requests to be intercepted and replayed from a cassette, demonstrating how to create a MockSocket and construct an HTTPResponse.

```python
from http.client import HTTPResponse
from io import BytesIO
from urllib.error import URLError
from urllib.request import BaseHandler, Request, build_opener

from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionRequest,
    ResponseChunk,
)

class MockSocket:
    """Mock socket for HTTPResponse construction."""
    def __init__(self, data: bytes) -> None:
        self.data = data

    def makefile(self, _mode: str = "rb", *_args, **_kwargs) -> BytesIO:
        return BytesIO(self.data)

class InterpositionHandler(BaseHandler):
    """urllib handler that delegates to Interposition Broker."""

    def __init__(self, broker: Broker) -> None:
        self.broker = broker

    def default_open(self, req: Request) -> HTTPResponse:
        # Convert urllib Request to InteractionRequest
        inter_req = InteractionRequest(
            protocol="http",
            action=req.get_method() or "GET",
            target=req.full_url,
            headers=(),
            body=req.data if isinstance(req.data, bytes) else b"",
        )

        try:
            response_gen = self.broker.replay(inter_req)
            full_content = b"".join(chunk.data for chunk in response_gen)
        except Exception as e:
            raise URLError(reason=str(e)) from e

        # Construct raw HTTP response
        raw_response = (
            b"HTTP/1.1 200 OK\r\n"
            b"Content-Type: application/json\r\n"
            f"Content-Length: {len(full_content)}\r\n".encode()
            + b"\r\n" + full_content
        )

        sock = MockSocket(raw_response)
        resp = HTTPResponse(sock, method=req.get_method(), url=req.full_url)
        resp.begin()
        return resp

```

--------------------------------

### Version Access

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Provides instructions on how to programmatically access the package version using the `__version__` attribute.

```APIDOC
## Version

Access the package version programmatically:

```python
from interposition import __version__

if __version__ < "0.2.0":
    print("Auto mode is not supported")
else:
    print("Auto mode is supported")
```
```

--------------------------------

### Implement Custom CassetteStore Protocol

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Defines a custom cassette store by implementing the CassetteStore protocol. This involves creating `load` and `save` methods for persisting cassette data. This allows for flexible storage solutions beyond simple file-based persistence.

```python
from interposition import CassetteStore

class MyCassetteStore:
    """Custom store implementation."""

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

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

--------------------------------

### Catch All Interposition Errors in Python

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

Demonstrates how to catch all interposition-specific exceptions using a single `InterpositionError` handler. This is useful for broad error management in applications using the interposition library.

```python
from interposition import InterpositionError

try:
    broker.replay(request)
except InterpositionError as e:
    print(f"Interposition error: {e}")
```

--------------------------------

### Interposition Error Handling in Python

Source: https://context7.com/osoekawaitlab/interposition/llms.txt

Demonstrates how to catch various Interposition-specific errors, such as InteractionNotFoundError, LiveResponderRequiredError, CassetteLoadError, and CassetteSaveError. It shows how to use a general InterpositionError catch-all for unified error handling.

```python
from interposition import (
    Broker,
    Cassette,
    InteractionRequest,
    InterpositionError,
    InteractionNotFoundError,
    LiveResponderRequiredError,
    CassetteLoadError,
    CassetteSaveError,
    JsonFileCassetteStore,
)
from pathlib import Path

# Catch all interposition errors
cassette = Cassette(interactions=())
broker = Broker(cassette=cassette)

request = InteractionRequest(
    protocol="http", action="GET", target="/missing", headers=(), body=b""
)

try:
    list(broker.replay(request))
except InterpositionError as e:
    print(f"Interposition error: {e}")

# InteractionNotFoundError - no matching interaction in replay mode
try:
    list(broker.replay(request))
except InteractionNotFoundError as e:
    print(f"Not recorded: {e.request.target}")

# LiveResponderRequiredError - record/auto mode without live_responder
try:
    Broker(cassette=cassette, mode="auto")  # No live_responder!
except LiveResponderRequiredError as e:
    print(f"live_responder required for {e.mode} mode")

# CassetteLoadError - file not found or corrupted
store = JsonFileCassetteStore(Path("missing.json"))
try:
    store.load()
except CassetteLoadError as e:
    print(f"Failed to load from {e.path}: {e.__cause__}")

# CassetteSaveError - I/O errors during save
store = JsonFileCassetteStore(Path("/readonly/cassette.json"))
try:
    store.save(cassette)
except CassetteSaveError as e:
    print(f"Failed to save to {e.path}: {e.__cause__}")

```

--------------------------------

### Define MQTT/PubSub Message Interaction Request

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This Python code snippet shows how to define an `InteractionRequest` for an MQTT or Pub/Sub message. It includes the protocol, action, target topic, and quality of service (QoS) headers.

```python
request = InteractionRequest(
    protocol="mqtt",
    action="subscribe",
    target="sensors/temp/room1",
    headers=(("qos", "1"),),
    body=b"",
)
# Replay returns stream of messages: b'24.5', b'24.6', ...
```

--------------------------------

### Pytest Fixture for Interposition Broker

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This Python code defines a pytest fixture named `cassette_broker`. It loads a cassette from a JSON file and initializes an Interposition Broker. This fixture can then be used in tests to replay recorded interactions.

```python
import pytest
from interposition import Broker, Cassette, InteractionRequest

@pytest.fixture
def cassette_broker():
    # Load cassette from a JSON file (or create one programmatically)
    with open("tests/fixtures/my_cassette.json", "rb") as f:
        cassette = Cassette.model_validate_json(f.read())
    return Broker(cassette)

def test_user_service(cassette_broker, monkeypatch):
    # 1. Create your adapter (mocking your actual client)
    def mock_fetch(url):
        request = InteractionRequest(
            protocol="http",
            action="GET",
            target=url,
            headers=(),
            body=b"",
        )
        # Delegate to Interposition
        chunks = list(cassette_broker.replay(request))
        return chunks[0].data

    # 2. Inject the adapter
    monkeypatch.setattr("my_app.client.fetch", mock_fetch)

    # 3. Run your test
    from my_app import get_user_name
    assert get_user_name(42) == "Alice"
```

--------------------------------

### Define SQL Database Interaction Request

Source: https://github.com/osoekawaitlab/interposition/blob/main/README.md

This Python code snippet demonstrates how to define an `InteractionRequest` for a SQL database query. It specifies the protocol, action, target, headers, and the SQL query body.

```python
request = InteractionRequest(
    protocol="postgres",
    action="SELECT",
    target="users_table",
    headers=(),
    body=b"SELECT id, name FROM users WHERE id = 42",
)
# Replay returns: b'[(42, "Alice")]'
```