Skip to main content
Below describes an example workflow of API calls that can be made in order to submit a Bulk Enhance request and retrieve results using the Ping.Data API. The following is a specific example; there are more options available to the user that are documented in the Bulk Enhance pages linked below under Ping.Data. The code blocks allow the user to select from Python or cURL (via terminal) use cases. The sample 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!

1. Start Bulk Enhance Job

The first API call to make is Start Bulk Enhance Job. This endpoint allows a user to submit a batch of locations to Ping for enhancement. Example code:
enhance_multiple_locations.py
import time
from pathlib import Path
import requests
import os

# authentication token that allows you to make requests to the API
API_KEY = os.environ.get('PING_DATA_AUTH_TOKEN')
headers = {"Authorization": f"Token {API_KEY}"}

# example batch payload (simple two-location example)
payload = {
    "locations": [
      {
        "id": "item-1",
        "address": "123 main st, miami, fl",
        "sources": ["PG", "PH"],
      },
      {
        "id": "item-2",
        "address": "456 elm st, los angeles, ca",
        "sources": ["PG", "PH"],
      }
    ],    
    "sources": ["PG", "PH"],
}

API_BASE = "https://data-api.sovfixer.com/api/v1"
# 1. API URL for Start Bulk Enhance Job: https://data-api.sovfixer.com/api/v1/bulk_enhance
start_job_url = f"{API_BASE}/bulk_enhance"
# API response
start_job_response = requests.post(start_job_url, json=payload, headers=headers)

# check response status and record the job ID
if start_job_response.status_code in (200, 201):
  	jobid = start_job_response.json()["id"]
	  print(start_job_response.json())
else:
	raise RuntimeError

## ...
Example response:
Python output
{'id': '3c90c3cc-0d44-4b50-8888-8dd25736052a', 'message': 'OK'}

2. Check Bulk Enhance Job Status

The second API call to make is Check Bulk Enhance Job Status. This endpoint allows a user to check the status of a previously submitted Bulk Enhance job using the id returned from the first step. The user should poll this endpoint until the job status is either COMPLETE or FAILED. Example code:
enhance_multiple_locations.py
## ...

# 2. API URL for Check Bulk Enhance Job Status: https://data-api.sovfixer.com/api/v1/bulk_enhance/{id}
check_job_url = f"{API_BASE}/bulk_enhance/{jobid}"
check_job_response = requests.get(check_job_url, headers=headers)
# ensure response is in a good state
assert check_job_response.status_code in (200, 201)
check_job_json = check_job_response.json()
# Poll every three seconds for job completion
while check_job_json.get("request", {}).get("status") not in ("COMPLETE", "FAILED"):
	  print(check_job_json)
	  print("Waiting 3 seconds to request again...")
	  time.sleep(3)
	  check_job_response = requests.get(check_job_url, headers=headers)
	  # ensure response is in a good state
	  assert check_job_response.status_code in (200, 201)
	  check_job_json = check_job_response.json()

print(check_job_json)

# record filenames and urls of the processed outputs
outputs = []
for output in check_job_json.get("result", {}).get("outputs", []):
    outputs.append(
		  	{
			  	"filename": output.get("filename"), 
				  "url": output.get("url")
			  }
		  )

## ...
Example (truncated) response:
Python output
{'request': {'completed_at': '2026-02-02T19:18:52.624Z',
             'num_canceled': 0,
             'num_completed': 2,
             'num_problems': 2,
             'num_requested': 2,
             'progress_started_at': '2026-02-02T19:18:49.264Z',
             'requested_at': '2026-02-02T19:18:48.842Z',
             'status': 'COMPLETE'},
 'result': {'message': 'Success.',
            'outputs': [{'description': 'Data file 1',
                         'filename': 'output-data-3c90c3cc-0d44-4b50-8888-8dd25736052a.json',
                         'url': 'https://data-api.sovfixer.com/api/v1/bulk_enhance/3c90c3cc-0d44-4b50-8888-8dd25736052a/output/output-data-3c90c3cc-0d44-4b50-8888-8dd25736052a.json'}],
            'sources': [{'avg_fetch_duration': 0.121,
                         'max_fetch_duration': 0.152,
                         'min_fetch_duration': 0.09,
                         'num_fails': 0,
                         'num_fetched': 2,
                         'num_successes': 2,
                         'source': 'PG',
                         'total_fetch_duration': 0.242},
                         'total_fetch_duration': 0.09},
                        {'avg_fetch_duration': 0.0795,
                         'max_fetch_duration': 0.08,
                         'min_fetch_duration': 0.079,
                         'num_fails': 0,
                         'num_fetched': 2,
                         'num_successes': 2,
                         'source': 'PGPRE',
                         'total_fetch_duration': 0.159},
                        {'avg_fetch_duration': None,
                         'max_fetch_duration': None,
                         'min_fetch_duration': None,
                         'num_fails': 0,
                         'num_fetched': 2,
                         'num_successes': 2,
                         'source': 'PH',
                         'total_fetch_duration': 0.0}],
            'status': 'SUCCESS',
            'total_processing_time': 3.782
            }
  }

3. Fetch Bulk Enhance Job Result Data

The third API call is Fetch Bulk Enhance Job Result Data. This endpoint allows a user to download the resultant file for a completed Bulk Enhance job using the id and filename returned from the previous step. Example code:
enhance_multiple_locations.py
## ...
# 3. API URL for Fetch Bulk Enhance Job Result Data: https://data-api.sovfixer.com/api/v1/bulk_enhance/{id}/output/{filename}
for output in outputs:
	fetch_outputs_response = requests.get(output["url"], headers=headers)
	# ensure response is in a good state
	assert fetch_outputs_response.status_code in (200, 201)
	print(f"saving {output['filename']}")
  # save to workflow_example_results directory
	out_dir = Path("workflow_example_results")
	out_dir.mkdir(exist_ok=True)
	with open(out_dir / output["filename"], "wb") as outfile:
		outfile.write(fetch_outputs_response.content)    
	print(f"saved {output['filename']}")
Example response:
Python output
saving bulk_enhance_3c90c3cc-results.json
saved bulk_enhance_3c90c3cc-results.json

Python Demo

Download a ready-to-run Python demo of this workflow here:                                   Download Python Script here
enhance_multiple_locations.py
import time
from pathlib import Path
import requests
import os

# authentication token that allows you to make requests to the API
API_KEY = os.environ.get('PING_DATA_AUTH_TOKEN')
headers = {"Authorization": f"Token {API_KEY}"}

# example batch payload (simple two-location example)
payload = {
  "locations": [
    {
      "id": "item-1",
      "address": "123 main st, miami, fl",
      "sources": ["PG", "PH"],
    },
    {
      "id": "item-2",
      "address": "456 elm st, los angeles, ca",
      "sources": ["PG", "PH"],
    }
  ],    
  "sources": ["PG", "PH"],
}

API_BASE = "https://data-api.sovfixer.com/api/v1"
# 1. API URL for Start Bulk Enhance Job: https://data-api.sovfixer.com/api/v1/bulk_enhance
start_job_url = f"{API_BASE}/bulk_enhance"
# API response
start_job_response = requests.post(start_job_url, json=payload, headers=headers)

# check response status and record the job ID
if start_job_response.status_code in (200, 201):
	jobid = start_job_response.json()["id"]
	print(start_job_response.json())

else:
	raise RuntimeError
	
## ...

# 2. API URL for Check Bulk Enhance Job Status: https://data-api.sovfixer.com/api/v1/bulk_enhance/{id}
check_job_url = f"{API_BASE}/bulk_enhance/{jobid}"
check_job_response = requests.get(check_job_url, headers=headers)
# ensure response is in a good state
assert check_job_response.status_code in (200, 201)
check_job_json = check_job_response.json()
# Poll every three seconds for job completion
while check_job_json.get("request", {}).get("status") not in ("COMPLETE", "FAILED"):
	print(check_job_json)
	print("Waiting 3 seconds to request again...")
	time.sleep(3)
	check_job_response = requests.get(check_job_url, headers=headers)
	# ensure response is in a good state
	assert check_job_response.status_code in (200, 201)
	check_job_json = check_job_response.json()

print(check_job_json)

# record filenames and urls of the processed outputs
outputs = []
for output in check_job_json.get("result", {}).get("outputs", []):
	outputs.append(
			{
				"filename": output.get("filename"), 
				"url": output.get("url")
			}
		)

## ...
# 3. API URL for Fetch Bulk Enhance Job Result Data: https://data-api.sovfixer.com/api/v1/bulk_enhance/{id}/output/{filename}
for output in outputs:
	fetch_outputs_response = requests.get(output["url"], headers=headers)
	# ensure response is in a good state
	assert fetch_outputs_response.status_code in (200, 201)
	print(f"saving {output['filename']}")
	# save JSON outputs as JSON; others as binary
	out_dir = Path("workflow_example_results")
	out_dir.mkdir(exist_ok=True)
	with open(out_dir / output["filename"], "wb") as outfile:
		outfile.write(fetch_outputs_response.content)    
	print(f"saved {output['filename']}")