ranglerpy 0.1.5

Python SDK for the Rangler API.

ranglerpy

ranglerpy is a Python SDK for the Rangler API.

It is designed around the actual Rangler product shape:

• polling-first event consumption
• optional webhook verification helpers

The goal is to make Rangler easy to integrate whether you want to:

• poll event feeds with a cursor
• consume companies and filings directly
• create Connect link tokens and read customer portfolio data
• verify Rangler webhook signatures in a receiver

It is designed to give you one integration surface across:

• polling feeds
• Connect data-plane workflows
• webhook receivers

Installation

With uv:

uv add ranglerpy

With pip:

pip install ranglerpy

For local development:

pip install -e .[dev]

For release builds:

python scripts/update_version.py 0.1.5
python -m unittest discover -s tests -v
uv build --no-sources
export UV_PUBLISH_TOKEN=pypi-your-project-token
uv publish --token "$UV_PUBLISH_TOKEN" dist/*

For local checks with the PyPA tools:

pip install -e .[build]
python -m build
python -m twine check dist/*

Quick start

Data plane: polling events

from ranglerpy import InMemoryCursorStore, PollingConsumer, RanglerClient

client = RanglerClient(
    api_key="rgl_test_your_api_key",
    environment="sandbox",
)

consumer = PollingConsumer(
    client.v1.events,
    cursor_store=InMemoryCursorStore(),
    stream="market-wide-filings",
)

for event in consumer.poll(event_types=["filing.new"], limit=100):
    print(event.id, event.type, event.occurred_at.isoformat())

Async polling

from ranglerpy import AsyncPollingConsumer, AsyncRanglerClient, FileCursorStore

async with AsyncRanglerClient(
    api_key="rgl_test_your_api_key",
    environment="sandbox",
) as client:
    consumer = AsyncPollingConsumer(
        client.v1.events,
        cursor_store=FileCursorStore(".rangler-cursors.json"),
        stream="issuer-monitoring",
    )

    for event in await consumer.poll(
        limit=100,
        event_types=["filing.new"],
    ):
        print(event.id, event.type)

Data plane: Connect

from ranglerpy import RanglerClient

client = RanglerClient(
    api_key="rgl_test_your_api_key",
    environment="sandbox",
)

link_token = client.connect.create_link_token(
    client_user_id="customer_123",
    allowed_origins=["https://app.example.com"],
    products=["accounts", "positions", "transactions"],
    idempotency_key="customer_123-connect-link",
)

connections = client.connect.list_connections(client_user_id="customer_123")
portfolio = client.connect.get_portfolio(client_user_id="customer_123")

print(link_token["link_token"], len(connections["data"]), portfolio["totals"])

Webhook verification

from ranglerpy import InMemoryIdempotencyStore, Webhook

idempotency_store = InMemoryIdempotencyStore()

event = Webhook.construct_event(
    headers=headers,
    raw_body=raw_body,
    secret=webhook_secret,
    idempotency_store=idempotency_store,
)

print(event.type, event.display.title, event.data.object.id)

FastAPI webhook handler

from fastapi import FastAPI, HTTPException, Request

from ranglerpy import DuplicateEventError, InMemoryIdempotencyStore, InvalidSignatureError, Webhook

app = FastAPI()
store = InMemoryIdempotencyStore()


@app.post("/rangler/webhooks")
async def rangler_webhook(request: Request):
    raw_body = await request.body()

    try:
        event = Webhook.construct_event(
            headers=request.headers,
            raw_body=raw_body,
            secret="whsec_your_secret",
            idempotency_store=store,
        )
    except InvalidSignatureError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except DuplicateEventError:
        return {"duplicate": True}

    return {"received": event.id}

Flask webhook handler

from flask import Flask, jsonify, request

from ranglerpy import DuplicateEventError, InMemoryIdempotencyStore, InvalidSignatureError, Webhook

app = Flask(__name__)
store = InMemoryIdempotencyStore()


@app.post("/rangler/webhooks")
def rangler_webhook():
    raw_body = request.get_data()

    try:
        event = Webhook.construct_event(
            headers=request.headers,
            raw_body=raw_body,
            secret="whsec_your_secret",
            idempotency_store=store,
        )
    except InvalidSignatureError as exc:
        return jsonify({"error": str(exc)}), 400
    except DuplicateEventError:
        return jsonify({"duplicate": True}), 200

    return jsonify({"received": event.id})

Authentication

ranglerpy uses X-API-Key authentication for public API operations.

Current SDK scope

Data plane

• companies
• Connect accounts, positions, transactions, portfolio, and link tokens
• filings
• funds
• event feeds

Helpers

• cursor-based event iteration
• Stripe-style client.v1.* namespace
• list responses with data
• dynamic object access for webhook/API payloads, e.g. event.data.object.id
• polling consumer with checkpoint persistence
• Rangler webhook signature verification
• Webhook.construct_event(...) for webhook receivers
• in-memory idempotency helper for duplicate webhook handling
• async client support

Example scripts

The repo includes runnable examples in examples/:

• poll_market_events.py for market-wide event polling
• poll_company_events.py for issuer-scoped polling
• poll_fund_events.py for fund-scoped polling
• fastapi_webhook_receiver.py for a minimal webhook receiver

These are intentionally small. They are meant to be copied into a real worker, cron job, or receiver and then adapted to your own storage and job queue.

Notes

This first version is intentionally thin. It aims to give Rangler customers one integration surface across polling and webhooks instead of pushing everyone straight into raw HTTP and receiver boilerplate.

You can override base_url explicitly if you need to point the SDK at a different Rangler environment.

The package version is tracked in VERSION, pyproject.toml, and src/ranglerpy/_version.py. The API contract version is tracked separately in src/ranglerpy/_api_version.py.