Monitor SOV Activity

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 workflow only observes activity. To submit parsing jobs and updates from the same loop that tracks them, see Build an SOV Pipeline. 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:

Poll List Historical SOVs for new activity using a cursor.
Distinguish SOVs from updates (SUDs, for SOV Update Data) in the response.
Fetch full details using Get Historical SOV.

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. Storing the full HistoryItem
# models the state a real integration would keep, e.g. to compare revisions
# as they land.
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. Keep the prior cursor if an empty
    # page omits or nulls cursor_id, so we resume after the last record we saw
    # rather than replaying from `start`.
    last_cursor_id = data.get("cursor_id") or last_cursor_id

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

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

# First poll: scope by start.
curl --request GET \
    --header "Authorization: Token <your_api_key_here>" \
    --url "https://api.sovfixer.com/api/v1/sov/history?start=20260526000000&page_size=50"

# Subsequent polls: pass the cursor_id from the previous response.
curl --request GET \
    --header "Authorization: Token <your_api_key_here>" \
    --url "https://api.sovfixer.com/api/v1/sov/history?cursor_id={cursor_id}&page_size=50"

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())

curl --request GET \
    --header "Authorization: Token <your_api_key_here>" \
    --url "https://api.sovfixer.com/api/v1/sov/history/{id}"

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. Storing the full HistoryItem
# models the state a real integration would keep, e.g. to compare revisions
# as they land.
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.get("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. Keep the prior cursor if an empty
    # page omits or nulls cursor_id, so we resume after the last record we saw
    # rather than replaying from `start`.
    last_cursor_id = data.get("cursor_id") or last_cursor_id

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

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

Overview

Ping.Extraction

Ping.Data

Ping.Vision

1. Poll List Historical SOVs

2. Get Historical SOV Details

Python Demo

​1. Poll List Historical SOVs

​2. Get Historical SOV Details

​Python Demo

1. Poll List Historical SOVs

2. Get Historical SOV Details

Python Demo