Overview

Quickstart Code Example

Run an extraction pipeline over a manifest — in Python

The Extract API runs a named pipeline over a set of documents listed in a manifest in your own S3 bucket. Like the Prep API it reads and writes your storage directly, so a job needs storage credentials as well as your API key.

$export EXTRACT_API_KEY='...'
$export AWS_ACCESS_KEY_ID='...'
$export AWS_SECRET_ACCESS_KEY='...'

Create a session

Authentication is an API key in the X-API-Key header.

1import os
2import requests
3
4BASE = "https://extraction-api.emtelligent.com"
5
6session = requests.Session()
7session.headers["X-API-Key"] = os.environ["EXTRACT_API_KEY"]

Find a pipeline

Pipelines are named, and which ones exist depends on your deployment — so ask rather than hard-coding a name.

1pipelines = session.get(f"{BASE}/api/v1/pipelines").json()
2print(pipelines)

Each pipeline takes its own parameters, which you can inspect before submitting:

1params = session.get(
2    f"{BASE}/api/v1/pipelines/core_clinical/pipeline_params"
3).json()

Create a job

manifest_key is the object key of the manifest within your input bucket, and aws_region is the region that bucket lives in.

No bucket name appears in the request. Your input and output buckets are fixed during onboarding and resolved from your API key, so a job names only the key inside them — the credentials you send are what lets the service read and write those buckets on your behalf.

1response = session.post(
2    f"{BASE}/api/v1/jobs",
3    json={
4        "manifest_key": "manifests/batch-001.json",
5        "pipeline_name": "core_clinical",
6        "aws_region": "us-east-1",
7        "user_credentials": {
8            "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
9            "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
10        },
11        "job_name": "batch-001",
12    },
13)
14response.raise_for_status()
15job_id = response.json()["job_id"]

These credentials are used by the service to read your documents and write your results. Scope them to the buckets this job needs.

Wait for it to finish

A job moves through queued, validating, ingesting and running before it reaches completed, failed or cancelled; it can also sit at paused. Collecting adds three more — see below. Once the retention window has passed the job reports output_expired: it succeeded, but its output can no longer be collected, so treat that as terminal too or a polling loop will never exit.

1import time
2
3TERMINAL = {"completed", "failed", "cancelled", "output_expired"}
4
5while True:
6    status = session.get(f"{BASE}/api/v1/jobs/{job_id}/status").json()
7    if status["job_status"] in TERMINAL:
8        break
9    time.sleep(10)
10
11if status["job_status"] != "completed":
12    raise RuntimeError(f"job {job_id} ended as {status['job_status']}")

Collect the output

Collecting is a transfer, not a download: the service copies the results into your bucket under output_prefix, with the job id appended, so it needs the credentials and region again.

1response = session.post(
2    f"{BASE}/api/v1/jobs/{job_id}/collect",
3    json={
4        "output_prefix": "extract-output",
5        "aws_region": "us-east-1",
6        "user_credentials": {
7            "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
8            "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
9        },
10    },
11)
12response.raise_for_status()

The job then reports transferring, and finishes at transfer_completed or transfer_failed. Your results land in your output bucket — again the one from onboarding, not named here — under extract-output/<job_id>/.

The whole thing

extract_quickstart.py
1import os
2import time
3
4import requests
5
6BASE = "https://extraction-api.emtelligent.com"
7REGION = "us-east-1"
8
9CREDENTIALS = {
10    "aws_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
11    "aws_secret_access_key": os.environ["AWS_SECRET_ACCESS_KEY"],
12}
13
14session = requests.Session()
15session.headers["X-API-Key"] = os.environ["EXTRACT_API_KEY"]
16
17
18def wait_for(job_id, done, every=10):
19    """Poll a job until its status is in `done`, then return that status."""
20    while True:
21        status = session.get(f"{BASE}/api/v1/jobs/{job_id}/status").json()
22        if status["job_status"] in done:
23            return status
24        time.sleep(every)
25
26
27def extract(manifest_key, pipeline_name, output_prefix, job_name=None):
28    """Run a pipeline over a manifest and copy the output to your bucket."""
29    response = session.post(
30        f"{BASE}/api/v1/jobs",
31        json={
32            "manifest_key": manifest_key,
33            "pipeline_name": pipeline_name,
34            "aws_region": REGION,
35            "user_credentials": CREDENTIALS,
36            "job_name": job_name,
37        },
38    )
39    response.raise_for_status()
40    job_id = response.json()["job_id"]
41
42    # `output_expired` is terminal as well: the job succeeded long enough ago that
43    # its output is gone, and waiting for `completed` would never return.
44    status = wait_for(job_id, {"completed", "failed", "cancelled", "output_expired"})
45    if status["job_status"] != "completed":
46        raise RuntimeError(f"job {job_id} ended as {status['job_status']}")
47
48    response = session.post(
49        f"{BASE}/api/v1/jobs/{job_id}/collect",
50        json={
51            "output_prefix": output_prefix,
52            "aws_region": REGION,
53            "user_credentials": CREDENTIALS,
54        },
55    )
56    response.raise_for_status()
57
58    status = wait_for(job_id, {"transfer_completed", "transfer_failed"})
59    if status["job_status"] != "transfer_completed":
60        raise RuntimeError(f"collecting job {job_id} ended as {status['job_status']}")
61
62    return f"{output_prefix}/{job_id}/"
63
64
65if __name__ == "__main__":
66    print(extract("manifests/batch-001.json", "core_clinical",
67                  "extract-output", job_name="batch-001"))

Next

API Reference

Every operation, including cancel and job listing.

Output reference

What the extracted data looks like, field by field.