Python SDK

The PulseRoute SDK provides a lightweight client for integrating routing decisions into your payment service.

Installation

pip install -r requirements.txt

The SDK is in src/sdk/client.py — no separate package needed.

Quick Start

Direct Mode (embedded engine)

For services that run PulseRoute in-process:

from src.sdk.client import PulseRouteClient
from src.models import HealthConfig

client = PulseRouteClient(
    config=HealthConfig(evaluation_interval_seconds=5),
)
client.start()

# Get routing decision
decision = client.get_route(
    country="US",
    currency="USD",
    payment_method="card",
    card_type="visa",
    amount=99.99,
)

if decision:
    processor = decision.processor_id  # "stripe"
    # ... process payment with this processor ...

    # Report outcome
    client.report_outcome(
        rule_id="us_cards",
        processor_id=processor,
        success=True,
        latency_ms=120.0,
    )

client.stop()

Remote Mode (API client)

For services that connect to a PulseRoute API server:

client = PulseRouteClient(
    base_url="http://pulseroute:8080",
    api_key="your-api-key",
)
client.start()

decision = client.get_route(country="US", currency="USD")

Parameters

get_route()

decision = client.get_route(
    country="US",              # Required: ISO 3166-1 alpha-2
    currency="USD",            # Required: ISO 4217
    payment_method="card",     # Default: "card"
    card_type="visa",          # Default: "*" (any)
    transaction_type="one_time",  # Default: "one_time"
    amount=99.99,              # Optional: auto-derives amount_tier
    amount_tier=None,          # Optional: override auto-derivation
)

Amount tier auto-derivation:

Amount Range	Tier
$0 - $10	micro
$10 - $1,000	standard
$1,000 - $10,000	high_value
$10,000+	premium

report_outcome()

client.report_outcome(
    rule_id="us_cards",
    processor_id="stripe",
    success=True,
    latency_ms=120.0,
    error_code=None,         # Optional: processor error code
)

Features

Outcome Buffering

Outcomes are buffered and flushed in batches to avoid per-transaction overhead:

client = PulseRouteClient(
    config=config,
    buffer_size=100,        # Flush after 100 outcomes
    flush_interval=5.0,     # Or every 5 seconds
)

Decision Caching

When the engine or API is temporarily unavailable, the SDK falls back to cached decisions:

client = PulseRouteClient(
    config=config,
    enable_cache=True,  # Default: True
)

# First call: hits engine, caches result
decision = client.get_route(country="US", currency="USD")

# If engine goes down, returns cached decision with source=FALLBACK
decision = client.get_route(country="US", currency="USD")
assert decision.decision_source == DecisionSource.FALLBACK

Format-Agnostic Adapters

Different processors return different response formats. Use adapters to normalize them:

from src.sdk.adapters import AdapterRegistry

registry = AdapterRegistry()

# Register a custom adapter for a processor
registry.register("stripe", {
    "success_field": "status",
    "success_value": "succeeded",
    "latency_field": "processing_time_ms",
    "error_field": "error.code",
})

client = PulseRouteClient(
    config=config,
    adapter_registry=registry,
)

# Report raw processor response — adapter normalizes it
client.report_outcome_raw(
    rule_id="us_cards",
    processor_id="stripe",
    raw_response={
        "status": "succeeded",
        "processing_time_ms": 120,
    },
)

Client Configuration

client = PulseRouteClient(
    # Mode selection (pick one)
    config=HealthConfig(),          # Direct mode: embedded engine
    base_url="http://host:8080",    # Remote mode: API client
    api_key="key",                  # API authentication

    # Buffering
    buffer_size=100,                # Outcome buffer size
    flush_interval=5.0,             # Auto-flush interval (seconds)

    # Caching
    enable_cache=True,              # Decision cache enabled

    # Remote mode settings
    http_timeout=5.0,               # HTTP request timeout
)

Lifecycle

client = PulseRouteClient(config=config)

# Start background evaluation + flush threads
client.start()

# ... use client ...

# Stop gracefully (flushes remaining outcomes)
client.stop()