Extract

documents.extract pulls structured data from a document into a JSON object you define. Supply an inline JSON Schema or the ID of a registered schema. Every declared field receives a status. not_found means NDI examined the document and did not find the value; it is not an HTTP or job failure.

Basic usage

from ndi_sdk import NdiClient

schema = {
    "type": "object",
    "properties": {
        "invoice_number": {
            "type": "string",
            "description": "Invoice identifier shown near the top",
        },
        "total": {
            "type": "number",
            "description": "Final amount due",
        },
    },
    "required": ["invoice_number", "total"],
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    queued = client.documents.extract(
        upload,
        json_schema=schema,
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: succeeded Job; job.result is an ExtractResult

    print(job.result.data)
    for field in job.result.fields:
        print(field.path, field.status, field.value)

Method signature

def extract(
    source: DocumentSource | UploadResponse,
    *,
    json_schema: JsonSchema | None = None,
    schema_id: str | None = None,
    instructions: str | None = None,
    page_ranges: list[PageRange] | None = None,
    parse_mode: Literal["low", "medium", "high"] = "low",
    citations: ExtractCitationOptions | None = None,
    delivery: Literal["complete", "progressive"] = "complete",
    wait_seconds: int = 0,
    idempotency_key: str | None = None,
) -> Job

Parameters

Parameter	Type	Required	Description
`source`	`DocumentSource` or `UploadResponse`	Yes	Staged upload, URL, workspace file, or successful parse job
`json_schema`	JSON Schema object	One of `json_schema` or `schema_id`	Inline schema describing the result. Do not send both
`schema_id`	`str`	One of `json_schema` or `schema_id`	Registered schema ID (`sch_...`). Do not send both
`instructions`	`str`	No	Additional guidance for filling declared fields
`page_ranges`	`list[PageRange]`	No	One-based inclusive ranges. Unavailable with a parse-result source
`parse_mode`	`"low"` / `"medium"` / `"high"`	No	Processing effort before extraction
`citations`	`ExtractCitationOptions`	No	Include source locations and quoted source text
`delivery`	`"complete"` / `"progressive"`	No	Publish values before highlights when progressive. Default complete
`wait_seconds`	`int`	No	Hold the submission request open for up to 300 seconds
`idempotency_key`	`str`	No	Override the key the SDK normally mints

Returns

A Job. After success, job.result is an ExtractResult:

Field	Description
`data`	Object shaped by the supplied schema. Missing values are `None`
`fields`	One status entry per schema leaf, including missing fields
`schema_id` / `schema_version`	Registered schema identity, or `None` for an inline schema
`converted_pdf_url`	Authenticated converted PDF for supported Office sources
`units`	Billable units
`warnings`	Non-fatal extraction notes
`enrichment_status`	`pending`, `complete`, or `degraded`; job status determines whether processing has finished

With delivery="progressive", read the job using client.jobs.get(job_id). A running job can already contain values with enrichment_status="pending". Continue reading the same job to receive the final highlights. jobs.wait waits for terminal status, so use individual reads when showing early values. Parquet extraction currently returns its result at completion.

Define the schema

Use the supported JSON Schema subset to describe the exact shape you need:

from ndi_sdk import NdiClient

schema = {
    "type": "object",
    "properties": {
        "supplier": {
            "type": "string",
            "description": "Legal name of the supplier",
        },
        "invoice_date": {
            "type": "string",
            "description": "Invoice issue date in YYYY-MM-DD format",
        },
        "currency": {
            "type": "string",
            "description": "ISO currency code stated on the invoice",
        },
        "total": {
            "type": "number",
            "description": "Final invoice total after tax",
        },
    },
    "required": ["supplier", "invoice_date", "currency", "total"],
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.documents.extract(upload, json_schema=schema)
    # Returns: Job

Write specific descriptions

Descriptions guide extraction and should distinguish similar values:

schema = {
    "type": "object",
    "properties": {
        "subtotal": {
            "type": "number",
            "description": "Amount before tax and shipping",
        },
        "tax": {
            "type": "number",
            "description": "Tax amount, not tax rate",
        },
        "amount_due": {
            "type": "number",
            "description": "Final outstanding balance after payments",
        },
    },
}

instructions can clarify how to fill these fields, but it cannot add fields that the schema does not declare:

from ndi_sdk import NdiClient

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.documents.extract(
        upload,
        json_schema=schema,
        instructions="Use the final totals section, not line-item amounts.",
    )
    # Returns: Job

Validate before extracting

Schema validation creates no job and spends no credits:

from ndi_sdk import NdiClient

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    check = client.documents.validate_extract_schema(json_schema=schema)
    # Returns: ExtractSchemaValidationResponse

    if not check.valid:
        for error in check.errors:
            print(error.path, error.message)

The response contains every discovered error and the normalized schema when validation succeeds.

Use a registered schema

from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.documents.extract(
        upload,
        schema_id="sch_01JZ8WQ4VK2P7T1J0Y6S4M3N2B",
    )
    # Returns: Job; ExtractResult includes schema_id and schema_version

Create and publish reusable schemas through the REST extraction-schema endpoints. The Python SDK does not currently expose a schema-registry namespace. See Extraction schemas.

Source options

Extract accepts staged uploads, HTTPS URLs, workspace files, and successful parse jobs. Reuse a parse when another operation already produced the content:

from ndi_sdk import NdiClient, ParseResultSource

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    parsed = client.jobs.wait(
        client.documents.parse(upload, wait_seconds=30).job_id
    )
    queued = client.documents.extract(
        ParseResultSource(job_id=parsed.job_id),
        json_schema=schema,
    )
    # Returns: Job; the earlier parse is reused

page_ranges is unavailable with ParseResultSource because the selected content is already fixed. Use page ranges when extracting from an original source:

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import PageRange

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.documents.extract(
        upload,
        json_schema=schema,
        page_ranges=[PageRange(start=1, end=5)],
        parse_mode="medium",
    )
    # Returns: Job for pages 1 through 5

Citations

Citations are enabled by default. Each field can carry the location and source text that supports its value:

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import ExtractCitationOptions

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.jobs.wait(
        client.documents.extract(
            upload,
            json_schema=schema,
            citations=ExtractCitationOptions(
                enabled=True,
                include_source_text=True,
            ),
            wait_seconds=30,
        ).job_id
    )
    # Returns: ExtractResult whose fields include citations

    for field in job.result.fields:
        for citation in field.citations:
            print(field.path, citation.location, citation.source_text)

Disable citations only when you do not need source verification:

from ndi_sdk import NdiClient
from ndi_sdk.models.document_ops import ExtractCitationOptions

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    upload = client.documents.create_upload("invoice.pdf")
    job = client.documents.extract(
        upload,
        json_schema=schema,
        citations=ExtractCitationOptions(enabled=False),
    )
    # Returns: Job; ExtractResult.fields have no citations

Response structure

# After client.jobs.wait returns a succeeded extract job:
result = ...

print(result.data["invoice_number"])
print(result.data["total"])

for field in result.fields:
    print(field.path)        # JSON Pointer, such as /invoice_number
    print(field.status)      # found, not_found, or ambiguous
    print(field.value)
    print(field.confidence)
    print(field.citations)

Use data for the schema-shaped answer and fields for verification and status handling. Missing values remain present as None; they are not silently dropped.

Extract response format

Review field statuses, confidence, citation locations, warnings, and registered schema metadata.

Async extract

import asyncio

from ndi_sdk import AsyncNdiClient


async def main() -> None:
    schema = {
        "type": "object",
        "properties": {"total": {"type": "number"}},
    }
    async with AsyncNdiClient() as client:
        upload = await client.documents.create_upload("invoice.pdf")
        queued = await client.documents.extract(
            upload,
            json_schema=schema,
            wait_seconds=30,
        )
        job = await client.jobs.wait(queued.job_id, timeout=600)
        # Returns: succeeded Job with an ExtractResult
        print(job.result.data)


asyncio.run(main())

Complete example

from ndi_sdk import NdiClient, ParseResultSource

schema = {
    "type": "object",
    "properties": {
        "invoice_number": {
            "type": "string",
            "description": "Supplier invoice identifier",
        },
        "supplier": {
            "type": "string",
            "description": "Legal supplier name",
        },
        "amount_due": {
            "type": "number",
            "description": "Final outstanding balance",
        },
    },
    "required": ["invoice_number", "supplier", "amount_due"],
}

with NdiClient() as client:
    check = client.documents.validate_extract_schema(json_schema=schema)
    if not check.valid:
        raise ValueError(check.errors)

    upload = client.documents.create_upload("invoice.pdf")
    parsed = client.jobs.wait(
        client.documents.parse(upload, wait_seconds=30).job_id,
        timeout=600,
    )
    queued = client.documents.extract(
        ParseResultSource(job_id=parsed.job_id),
        json_schema=schema,
        instructions="Read the final balance after credits and prior payments.",
        wait_seconds=30,
    )
    job = client.jobs.wait(queued.job_id, timeout=600)
    # Returns: ExtractResult with schema-shaped data and per-field evidence

    print(job.result.data)
    missing = [
        field.path
        for field in job.result.fields
        if field.status == "not_found"
    ]
    print("Missing:", missing)

Error handling

from ndi_sdk import InvalidRequestError, JobFailedError, NdiClient

schema = {
    "type": "object",
    "properties": {"total": {"type": "number"}},
}

with NdiClient() as client:
    try:
        queued = client.documents.extract(
            client.documents.create_upload("invoice.pdf"),
            json_schema=schema,
        )
        job = client.jobs.wait(queued.job_id, timeout=600)
        # Returns: succeeded Job, or raises for request/job failure
    except InvalidRequestError as exc:
        print(exc.code, exc.message)
    except JobFailedError as exc:
        print(exc.job.status, exc.job.error)

Client-side ValueError is raised when both json_schema and schema_id are set, or when neither is set.

Best practices

Validate first

Preflight a schema without creating a job or spending credits.

Describe each field

State which value to select, how to distinguish it, and the expected representation.

Get started

Core methods

Async and utilities

Basic usage

Method signature

Parameters

Returns

Define the schema

Write specific descriptions

Validate before extracting

Use a registered schema

Source options

Citations

Response structure

Extract response format

Async extract

Complete example

Error handling

Best practices

Validate first

Describe each field

Keep citations on

Inspect not_found

Next steps

​Basic usage

​Method signature

​Parameters

​Returns

​Define the schema

​Write specific descriptions

​Validate before extracting

​Use a registered schema

​Source options

​Citations

​Response structure

Extract response format

​Async extract

​Complete example

​Error handling

​Best practices

Validate first

Describe each field

Keep citations on

Inspect not_found

​Next steps

Basic usage

Method signature

Parameters

Returns

Define the schema

Write specific descriptions

Validate before extracting

Use a registered schema

Source options

Citations

Response structure

Async extract

Complete example

Error handling

Best practices

Next steps