Batch Processing

1import requests
2import time
3
4API_KEY = "YOUR_API_KEY"
5BASE_URL = "https://api.magellandata.io/v1"
6HEADERS = {"Authorization": f"Bearer {API_KEY}"}
7
8urls = ["https://example.com", "https://subsidiary.example.com", "https://other.example.com"]
9
10# 1. Create the batch
11r = requests.post(f"{BASE_URL}/batch", headers=HEADERS, json={"urls": urls})
12r.raise_for_status()
13batch_id = r.json()["batch_id"]
14print(f"Batch created: {batch_id}")
15
16# 2. Kick off processing for the output types you want
17r = requests.post(
18    f"{BASE_URL}/batch/{batch_id}/process",
19    headers=HEADERS,
20    json={"output_types": ["parent_companies", "corporate_families"]},
21)
22r.raise_for_status()
23runs = r.json()["runs"]
24
25# 3. Poll each run until it completes
26completed = {}
27pending = {run["run_id"]: run["output_type"] for run in runs}
28while pending:
29    for run_id in list(pending.keys()):
30        r = requests.get(f"{BASE_URL}/batch/runs/{run_id}", headers=HEADERS)
31        r.raise_for_status()
32        status = r.json()
33        if status["run_status"] == "completed":
34            completed[run_id] = status
35            del pending[run_id]
36            print(f"{status['output_type']}: completed, {status['price']} credits")
37        elif status["run_status"] == "failed":
38            print(f"{pending[run_id]}: FAILED")
39            del pending[run_id]
40    if pending:
41        time.sleep(5)
42
43# 4. Download each completed run — two phases:
44#    a) POST to charge credits and get a presigned S3 URL
45#    b) GET the URL to stream the CSV to disk
46for run_id, status in completed.items():
47    output_type = status["output_type"]
48
49    # 4a. Purchase + URL
50    r = requests.post(f"{BASE_URL}/batch/runs/{run_id}/download", headers=HEADERS)
51    if r.status_code == 402:
52        err = r.json()
53        print(f"Insufficient credits: have {err['available_credits']}, need {err['required_credits']}")
54        continue
55    r.raise_for_status()
56    download_url = r.json()["download_url"]
57
58    # 4b. Stream the CSV from S3.
59    #     Do NOT pass headers=HEADERS — the URL is self-signed and S3 will
60    #     reject extra Authorization headers.
61    csv_path = f"{output_type}.csv"
62    with requests.get(download_url, stream=True) as resp:
63        resp.raise_for_status()
64        with open(csv_path, "wb") as f:
65            for chunk in resp.iter_content(chunk_size=1024 * 1024):
66                f.write(chunk)
67    print(f"{output_type}: saved to {csv_path}")