> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pingintel.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Monitor SOV Activity

> Example workflow for tracking every new SOV and SOV Update across an account via the **Ping.Extraction** API

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](/workflows/ping-extraction/build-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](/ping-extraction/get-sov-data/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](/ping-extraction/get-sov-data/get-historical-sov).

### 1. Poll List Historical SOVs

Poll [List Historical SOVs](/ping-extraction/get-sov-data/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:**

<CodeGroup dropdown>
  ```python monitor_sov_activity.py lines icon='python' theme={null}
  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)
  ```

  ```shell cURL output icon='square-terminal' theme={null}
  # 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"
  ```
</CodeGroup>

**Example response:**

<CodeGroup dropdown>
  ```json JSON output lines icon='json' theme={null}
  {
    "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"
      },
      { "...": "..." }
    ]
  }
  ```
</CodeGroup>

### 2. Get Historical SOV Details

From inside the step 1 loop, call [Get Historical SOV](/ping-extraction/get-sov-data/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:**

<CodeGroup dropdown>
  ```python monitor_sov_activity.py lines icon='python' theme={null}
  details = requests.get(f"{HISTORICAL_SOVS_URL}/{record['id']}", headers=headers)
  details.raise_for_status()
  print("details:")
  pprint(details.json())
  ```

  ```shell cURL output icon='square-terminal' theme={null}
  curl --request GET \
      --header "Authorization: Token <your_api_key_here>" \
      --url "https://api.sovfixer.com/api/v1/sov/history/{id}"
  ```
</CodeGroup>

**Example response:**

<CodeGroup dropdown>
  ```json JSON output lines icon='json' theme={null}
  {
    "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"
        }
      ],
      "...": "..."
    }
  }
  ```
</CodeGroup>

### 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`.

                                

<a href="https://raw.githubusercontent.com/pingintel/pingintel-api/main/examples/ping-extraction/monitor_sov_activity.py" download>
  Download Python Script here
</a>

<CodeGroup dropdown>
  ```python monitor_sov_activity.py lines icon='python' theme={null}
  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)
  ```
</CodeGroup>