StubSmith Python SDK

Instrument outbound HTTP calls made by your Python application and forward them, request and response, to the StubSmith ingest service for capture, anonymization, and replay.

Supports both requests and httpx (sync and async). Sending is non-blocking and fire-and-forget: a background daemon thread drains a bounded queue; any failure (network, serialization, queue overflow) is silently discarded and never propagates to your application.

Anonymization / masking is applied server-side by the Go ingest service (ingest-go) - do not pre-mask data client-side.

Install

# with requests support
pip install "stubsmith[requests]"

# with httpx support
pip install "stubsmith[httpx]"

# both
pip install "stubsmith[requests,httpx]"

Quickstart

import stubsmith

# One-liner: instruments both requests and httpx (whichever is importable)
client = stubsmith.install(api_key="sk-your-project-key")

# Now every outbound call is captured automatically:
import requests
resp = requests.get("https://api.stripe.com/v1/charges")

import httpx
resp = httpx.get("https://api.example.com/users")

Or configure via environment variables and call install() with no arguments:

export STUBSMITH_URL="http://ingest:8081/v1/captures"
export STUBSMITH_API_KEY="sk-your-project-key"

import stubsmith
stubsmith.install()

Selective instrumentation

from stubsmith import StubSmith

client = StubSmith(url="...", api_key="sk-...")
client.instrument_requests()   # only requests
# client.instrument_httpx()    # only httpx

# Remove patches later:
client.uninstrument()

Configuration

An outage cannot block your application

Captures are masked on the calling thread (CPU only, no I/O) and handed to a bounded queue with a non-blocking put. A daemon thread does the HTTP POST and swallows every exception. If the ingest service is slow, unreachable or returning errors, your request path is unaffected: the queue fills, excess captures are dropped, and nothing propagates to your code.

Measured with the ingest host blackholed, so every POST runs to its timeout: a median of 0.32 ms and a p95 of 0.45 ms added to an instrumented call.

The one place an outage is visible is process exit, where an atexit hook waits for the queue to drain. That wait is capped at flush_timeout, and is abandoned as soon as a send fails, so an unreachable endpoint costs about a second rather than the full budget. Set STUBSMITH_FLUSH_TIMEOUT=0 in a serverless function or anywhere else exit latency is billed; captures still in the queue are discarded.

How it works

1. install() / instrument_requests() / instrument_httpx() monkey-patches the relevant HTTP client (idempotent; safe to call multiple times).
2. Each call is timed; request headers/body and response status/headers/body are captured without consuming streams - if you opened a streaming response the SDK skips the body rather than interfering.
3. Captures are placed on an in-process queue.Queue; a daemon thread drains it and POSTs to POST /v1/captures with Authorization: Bearer <api_key>.
4. On process exit an atexit handler flushes the queue (bounded timeout).

Fingerprint value discrimination

By default the SDK fingerprints each request by its body structure (key-paths), query parameter names, and Content-Type. Endpoints that multiplex operations via a body field (e.g. {"action": "login"} vs {"action": "delete_user"}) therefore produce a single fingerprint, which means a single review and a single privacy-rule set for both variants.

Enable value discrimination on action in the StubSmith UI or via the API and the SDK will automatically include that field's value in the hash:

# No SDK change required - configure via the UI or API, then the SDK picks up the
# new value paths on its next rules-sync poll (default: every 60 seconds).
import stubsmith
stubsmith.install(url="...", api_key="sk-...")

import requests
requests.post("https://api.example.com/rpc", json={"action": "login", "username": "alice"})
# → fingerprint A  (action=login)

requests.post("https://api.example.com/rpc", json={"action": "delete_user", "user_id": 42})
# → fingerprint B  (action=delete_user) - separate review, separate rules

See docs/fingerprint-value-discrimination.md for a full walkthrough including the hash mechanics, all three configuration methods, and the privacy guarantees.

Offline replay in tests

stubsmith.replay() serves recorded responses to your code's outbound HTTP calls. Inside the block no network call is made and the dependency does not need to be running.

import stubsmith

def test_charge_is_declined():
    with stubsmith.replay():
        with pytest.raises(CardDeclined):
            PaymentClient().charge(amount_cents=950_000, currency="USD")

Fetch the bundle once and commit it:

export STUBSMITH_API_KEY=<your project key>
stubsmith pull --out .stubsmith/bundle.json

python -m stubsmith pull does the same thing. It resolves the package from sys.path, so it runs a checkout without installing it, and it cannot pick up a stale installed copy in place of the one you are working on.

From then on the test suite needs no key, no network and no pull step - it reads the committed file. Refresh it when the recording should change: the upstream API's shape changed, you added a call the bundle does not cover, or you approved new fingerprints. Treat it like a lockfile or a golden file: an occasional, reviewed, committed change.

replay() finds the bundle without configuration - an explicit path, then $STUBSMITH_BUNDLE, then an upward search from the working directory that stops at the first directory containing .git or pyproject.toml. Pass a path or a dict to be explicit:

with stubsmith.replay("tests/data/bundle.json"):
    ...

Matching

A request is matched on (domain, method, path_template, fingerprint). The fingerprint covers body key-paths, query parameter names and the normalised content-type - not values, and not the host or path, which is why the other three parts of the key are needed. Every body-less GET shares one fingerprint.

Dynamic path segments are templated from the recording, so /api/users/4821 matches a stub recorded as /api/users/{id}.

When nothing matches

StubNotFound is raised with a diff of what was sent against the closest recording, naming the fields that differ. It never falls through to the network, so a test cannot silently start calling a real service.

A stub whose fingerprint has no recorded captures is reported as degraded by stubsmith pull and raises the same error at replay time, rather than serving an empty response.

See examples/fixtures-testing/ for a complete worked example: a real service, a client instrumented with the SDK, traffic captured and reviewed, and a test suite that passes with the service stopped.

Single-fixture helpers

stubsmith.testing handles individual fixture files rather than a whole bundle, for cases where you want one recorded exchange registered against responses:

pip install "stubsmith[testing]"

from stubsmith import testing

bundle = testing.load_bundle("fixtures/get_user.json")
testing.register_template(responses, bundle, base_url="http://api")

It also provides assert_request_matches_fixture and assert_body_schemas_match as contract guards. For a normal test suite replay() is the simpler path - it covers every recorded endpoint at once and needs no per-fixture registration.

Masking and placeholders

Values are masked in the SDK, before a capture is uploaded: the server never receives the originals. A field with no keep rule is replaced.

By default the replacement is a constant ("<masked>", 0, False) matching the original's type.

Setting STUBSMITH_MASK_SALT to any non-empty string switches on format-preserving placeholders for fields carrying a semantic type hint (email, uuid, iso8601, e164, iban, url, decimal_amount, integer_id, opaque_token, free_text). The replacement then has the same shape as the original - a parseable timestamp, an RFC 4122 UUID, an IBAN with a correct mod-97 checksum - so code that parses or validates these values still works against a recording. The same salt and value always produce the same placeholder, which preserves uniqueness and cross-field references. The salt never leaves the process.

Two types are never format-preserved, regardless of salt: currency_code and country_code. Booleans are refused for the same reason. Their domains are small enough that a keyed hash could be reversed with a lookup table. Use action: keep for those fields instead - they are rarely sensitive.

Development / tests

python3 -m venv .venv && . .venv/bin/activate
pip install -e '.[test]'
python -m pytest -q