Python SDK

regsn is the official Python client for the Regulatory Snapshot API. It is a small, single-file library with no third-party dependencies — it uses urllib from the Python standard library, which means it runs anywhere Python ≥ 3.8 runs without a requirements.txt headache.

This page is the full reference. If you’ve never made a request to the API before, start with the Quickstart.

Install

pip install regsn

Requires Python 3.8 or newer. There are no runtime dependencies.

Authenticate

The SDK reads REGSN_API_KEY from the environment by default:

export REGSN_API_KEY="regsn_live_aBcD1234eFgH5678iJkL9012mNoP3456"

from regsn import RegSn
client = RegSn()

Or pass the key explicitly (useful for multi-tenant code):

client = RegSn(api_key="regsn_live_…")

Other constructor arguments:

The four resources

client.scans       # POST/GET /v1/scans, GET /v1/scans/:id
client.snapshots   # GET /v1/snapshots, GET /v1/snapshots/:id (+ sub-resources)
client.exports     # POST/GET /v1/exports
client.usage       # GET /v1/usage

client.scans

create(...) -> dict

Run a scan, defaulting to sync mode with auto-polling.

result = client.scans.create(
    jurisdictions=["UK", "EU"],
    areas=["AML", "Sanctions"],
    horizon=12,
    engine="v4.5-beta",          # optional
    verificationMode="in-analyst",  # optional
)
 
# For a deterministic envelope shape, re-fetch the snapshot.
snapshot = client.snapshots.get(result["snapshot_id"])
print(snapshot["data"]["executive_narrative"])

Behaviour:

• Sends POST /v1/scans?mode=sync with an auto-generated Idempotency-Key.
• If the response is 200 (scan completed within the 120s ceiling), returns it directly — the envelope is flat under result["snapshot"].
• If the response is 202 (scan still running), the SDK polls until done (poll_until_done=True default) by calling wait() under the hood, then returns the full snapshot envelope under result["snapshot"]["data"] (because wait() re-fetches via GET /v1/snapshots/{id}).
• Either way, calling client.snapshots.get(result["snapshot_id"]) after the scan gives you the stable nested shape — that’s what the rest of the docs use.

Keyword arguments:

create_async(...) -> dict

Convenience for ?mode=async. Returns immediately with scan_id + status_url + stream_url.

job = client.scans.create_async(
    jurisdictions=["UK"], areas=["AML"], horizon=12,
)
print(job["scan_id"])

get(scan_id) -> dict

Returns GET /v1/scans/{id}.

s = client.scans.get(scan_id)
print(s["status"], s["progress"])

wait(scan_id, *, poll_s=5, timeout_s=720) -> dict

Polls until the scan reaches a terminal state, then fetches the full snapshot envelope.

result = client.scans.wait(scan_id, poll_s=5, timeout_s=600)

Raises RegSnError(code="scan_failed") if the scan ends in status: failed, or RegSnError(code="wait_timeout") if timeout_s elapses first.

client.snapshots

listing = client.snapshots.list(starred="true", limit=20)
full = client.snapshots.get(snapshot_id)
items = client.snapshots.get_items(snapshot_id)
trends = client.snapshots.get_trends(snapshot_id)
summary = client.snapshots.get_executive_summary(snapshot_id)
narrative = client.snapshots.get_executive_narrative(snapshot_id, language="en")
drift = client.snapshots.get_drift(snapshot_id)

list() accepts the same query params as GET /v1/snapshots — pass them as keyword args.

client.exports

job = client.exports.create(
    snapshot_id=snapshot_id,
    artifact_type="pdf",
    provider="internal-pdf",
    options={},
)
print(job["export_job_id"])
 
# Poll until done.
while True:
    status = client.exports.get(job["export_job_id"])
    if status["status"] in ("completed", "failed"):
        break
    time.sleep(5)

The (provider, artifact_type) matrix is documented on Exports.

client.usage

data = client.usage.get(group_by="day")
print(data["totals"])
for bucket in data["buckets"]:
    print(bucket["key"], bucket["cost_cents"])

All query params from GET /v1/usage are accepted as kwargs.

Errors

All API errors are raised as RegSnError:

from regsn import RegSn, RegSnError
 
client = RegSn()
try:
    client.scans.create(jurisdictions=["UK"], areas=["AML"], horizon=13)
except RegSnError as e:
    print(e.status)       # 422
    print(e.code)         # "validation_error"
    print(e.request_id)   # "req_…"
    print(e.body)         # the full problem-details body, parsed

Retries

The SDK retries automatically on:

• 5xx responses (3 attempts, exponential backoff with jitter)
• 429 responses (respects Retry-After, capped at 60 seconds)
• Network errors (URLError)

After 3 retries it raises RegSnError(code="retry_exceeded"). For non-retryable errors (4xx besides 429), the first response is raised immediately.

The idempotency key is preserved across retries — that’s the whole point. As long as the server received the original request, retries will replay the cached response instead of starting a new scan.

Sync wrapper pattern

The scans.create() default behaviour is the most common pattern: fire a scan, block until done, return the snapshot. If you’d rather drive a job queue:

# Producer: enqueue scans, don't wait.
job = client.scans.create_async(jurisdictions=["UK"], areas=["AML"], horizon=12)
redis.rpush("regsn:in-flight", job["scan_id"])
 
# Worker: pop scan_id, wait for it, persist the result.
scan_id = redis.blpop("regsn:in-flight")[1].decode()
result = client.scans.wait(scan_id)
# wait() re-fetches via GET /v1/snapshots/{id}, so the envelope is at result["snapshot"]["data"]
persist(result["snapshot"]["data"])

See also

• Scans — request body, modes, streaming.
• Snapshots — the data model the SDK returns.
• JavaScript SDK — same surface, different language.
• Errors — error codes you’ll see in RegSnError.code.
• Glossary — bearer key, idempotency key, request ID.