Skip to main content
This example describes an integration that tracks every SOV processed or updated in Ping.Extraction. The workflow described polls a single cursor-paginated endpoint which supplies all completed or failed items. This cursor-based approach is preferred over callbacks, because:
  • the client system doesn’t need to expose any endpoints to the public internet.
  • authentication is one-way
  • if a problem or outage occurs on either side, recovering the lost messages is simple: just back the cursor up and reprocess the missed section as desired.
  • the client only needs to poll a single endpoint, at any desired frequency (5 seconds or 5 days!), rather than tracking multiple in-flight requests to observe transactional status
The code blocks let you choose between Python and cURL. A runnable Python script is available for download at the bottom of the page. Try filling in the blanks (e.g., {id}) to go through the workflow using cURL. This workflow demonstrates how to:

​
1. Poll List Historical SOVs

Poll List Historical SOVs to retrieve a chronological list of SOVs and SUDs processed by the system. The endpoint is cursor-paginated: pass the cursor_id returned by the previous response to fetch only records added since the last poll. Treat the cursor as opaque — pass it through unchanged rather than parsing it. Save each returned id if you plan to fetch full details in step 2. Each result includes a record_type. ORIG indicates a newly parsed SOV; any other value indicates a later revision (an SUD). The accompanying revision field is 0 for the original record and increments with each update. status uses single-letter codes: C for complete and F for failed. Example code:
monitor_sov_activity.py
import os
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pprint import pprint

import requests

@dataclass(frozen=True)
class HistoryItem:
    id: str
    revision: int
    record_type: str
    is_data_ready: bool = False

# Auth token; generate one at https://auth.pingintel.com/account/api_keys/

API_KEY = os.environ.get("SOVFIXER_AUTH_TOKEN")
BASE_URL = "https://api.sovfixer.com/api/v1"
HISTORICAL_SOVS_URL = f"{BASE_URL}/sov/history"

headers = {"Authorization": f"Token {API_KEY}"}

# `start` isn't required. leaving it out/passing null starts at the beginning of time.
start = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y%m%d%H%M%S")

# `cursor` is None on the first poll; subsequent polls pass the value returned
# by the previous response to fetch only records added since.

last_cursor_id = None

# Track revisions already reported, keyed by id, so overlapping pages don't print duplicates.

ids_seen: dict[str, HistoryItem] = {}

def process_record(record):
    """Perform storage, actions, logging, etc. here."""
    # `record_type` "ORIG" is an original SOV. Any other value indicates a later revision (SUD).
    kind = "SOV" if record["record_type"] == "ORIG" else "SUD"
    print(f"--- New {kind} ---")
    pprint(record)


while True:
    params = {"cursor_id": last_cursor_id, "start": start, "page_size": 50}
    response = requests.get(HISTORICAL_SOVS_URL, headers=headers, params=params)
    response.raise_for_status()

    data = response.json()
    # Advance the cursor for the next poll.
    last_cursor_id = data["cursor_id"]

    for record in data["results"]:
        if record["id"] in ids_seen:
            continue
        ids_seen[record["id"]] = HistoryItem(
            id=record["id"],
            revision=record["revision"],
            record_type=record["record_type"],
            is_data_ready=record["is_data_ready"],
        )
        process_record(record)

    print("Waiting 30 seconds to poll again...")
    time.sleep(30)
Example response:
JSON output
{
  "cursor_id": "s-lo-ping-zk9q2x",
  "results": [
    {
      "client_ref": null,
      "completed_time": "2026-03-24T13:46:04.165482Z",
      "id": "s-lo-ping-9xkw3m",
      "incremental": false,
      "is_data_ready": false,
      "pingid": null,
      "record_type": "ORIG",
      "revision": 0,
      "sovid": "s-lo-ping-9xkw3m",
      "status": "C"
    },
    {
      "client_ref": null,
      "completed_time": "2026-03-24T14:02:17.831204Z",
      "id": "s-lo-ping-4j2qhr",
      "incremental": false,
      "is_data_ready": false,
      "pingid": null,
      "record_type": "ORIG",
      "revision": 0,
      "sovid": "s-lo-ping-4j2qhr",
      "status": "F"
    },
    {
      "client_ref": null,
      "completed_time": "2026-03-24T14:18:45.092716Z",
      "id": "s-lo-ping-vt7n6c",
      "incremental": false,
      "is_data_ready": false,
      "pingid": null,
      "record_type": "ORIG",
      "revision": 0,
      "sovid": "s-lo-ping-vt7n6c",
      "status": "C"
    },
    { "...": "..." }
  ]
}

​
2. Get Historical SOV Details

From inside the step 1 loop, call Get Historical SOV with a record’s id to retrieve its full processing details. The payload is wrapped under result and includes the source filename, row count, submission metadata, lineage fields (original_sovid and previous_sovid for tracing SUDs back to their parent SOV), and an outputs[] array with a download URL for each generated artifact. The demo script calls this for failed records (status F) to surface the error_message. E.g., https://api.sovfixer.com/api/v1/sov/history/s-lo-ping-fd2acv Example code:
monitor_sov_activity.py
details = requests.get(f"{HISTORICAL_SOVS_URL}/{record['id']}", headers=headers)
details.raise_for_status()
print("details:")
pprint(details.json())
Example response:
JSON output
{
  "result": {
    "sovid": "s-lo-ping-fd2acv",
    "status": "C",
    "document_type": "SOV",
    "filename": "example_sov.xlsx",
    "sheet_name": "Sheet1",
    "num_rows": 311,
    "created_time": "2026-03-25T19:53:27.180Z",
    "completed_time": "2026-03-25T19:54:06.303Z",
    "original_sovid": null,
    "previous_sovid": null,
    "client_ref": null,
    "error_message": null,
    "from_email": null,
    "to_email": null,
    "subject": null,
    "organization_name": "Acme Insurance",
    "team_name": "Acme Insurance",
    "outputs": [
      {
        "completed_time": "2026-03-25T19:54:01.507Z",
        "label": "DEBUGJSON",
        "output_format": "DEBUGJSON",
        "scrubbed_filename": "example_sov.debug.json",
        "url": "https://api.sovfixer.com/api/v1/sov/s-lo-ping-fd2acv/output/example_sov.debug.json"
      }
    ],
    "...": "..."
  }
}

​
Python Demo

A minimal runnable script that polls the history endpoint and prints each new SOV and SUD as it appears. Set SOVFIXER_AUTH_TOKEN and run with python monitor_sov_activity.py.                                  Download Python Script here
monitor_sov_activity.py
import os
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pprint import pprint

import requests

@dataclass(frozen=True)
class HistoryItem:
    id: str
    revision: int
    record_type: str
    is_data_ready: bool = False

# Auth token; generate one at https://auth.pingintel.com/account/api_keys/
API_KEY = os.environ.get("SOVFIXER_AUTH_TOKEN")
BASE_URL = "https://api.sovfixer.com/api/v1"
HISTORICAL_SOVS_URL = f"{BASE_URL}/sov/history"

headers = {"Authorization": f"Token {API_KEY}"}

# `start` isn't required. leaving it out/passing null starts at the beginning of time.
start = (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y%m%d%H%M%S")

# `cursor` is None on the first poll; subsequent polls pass the value returned
# by the previous response to fetch only records added since.

last_cursor_id = None

# Track revisions already reported, keyed by id, so overlapping pages don't print duplicates.

ids_seen: dict[str, HistoryItem] = {}

def process_record(record):
    """Perform storage, actions, logging, etc. here."""
    # `record_type` "ORIG" is an original SOV. Any other value indicates a later revision (SUD).
    kind = "SOV" if record["record_type"] == "ORIG" else "SUD"
    print(f"--- New {kind} ---")
    pprint(record)

    # Fetch full details for failed records to surface the error_message.
    if record["status"] == "F":
        details = requests.get(
            f"{HISTORICAL_SOVS_URL}/{record['id']}", headers=headers
        )
        details.raise_for_status()
        print("details:")
        pprint(details.json())


while True:
    params = {"cursor_id": last_cursor_id, "start": start, "page_size": 50}
    response = requests.get(HISTORICAL_SOVS_URL, headers=headers, params=params)
    response.raise_for_status()

    data = response.json()
    # Advance the cursor for the next poll.
    last_cursor_id = data["cursor_id"]

    for record in data["results"]:
        if record["id"] in ids_seen:
            continue
        ids_seen[record["id"]] = HistoryItem(
            id=record["id"],
            revision=record["revision"],
            record_type=record["record_type"],
            is_data_ready=record["is_data_ready"],
        )
        process_record(record)

    print("Waiting 30 seconds to poll again...")
    time.sleep(30)