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

# Job management

> Wait for, inspect, list, stream, cancel, and delete jobs

Slow NDI operations return a `Job`. You can wait for completion, read the
latest state, stream progress, cancel active work, or list past jobs through
`client.jobs`.

## Submit and wait

```python theme={"dark"}
from ndi_sdk import NdiClient

with NdiClient() as client:
    upload = client.documents.create_upload("report.pdf")
    queued = client.documents.parse(upload)
    job = client.jobs.wait(queued.job_id, timeout=600)
    print(job.status, job.result.markdown)
```

`jobs.wait` polls until the job is terminal. Its default timeout is 300
seconds. A timeout stops waiting but does not cancel the server-side job.

## Choose inline wait or polling

Job-creating methods accept `wait_seconds`. This asks the server to hold the
submission response open for up to 300 seconds:

```python theme={"dark"}
queued_or_finished = client.documents.parse(upload, wait_seconds=30)
```

The method returns a `Job` either way. Read its `status`; if it is still
`queued` or `running`, continue with `client.jobs.wait`.

```python theme={"dark"}
job = queued_or_finished
if not job.is_terminal:
    job = client.jobs.wait(job.job_id, timeout=600)
```

`wait_seconds` is a server-side hold-open window. The `timeout` passed to
`jobs.wait` is the SDK's total polling budget. Neither timeout cancels the job.

## Retrieve job status

Use `jobs.get` when you control polling or need the latest state once.

```python theme={"dark"}
from ndi_sdk import JobStatus, NdiClient

with NdiClient() as client:
    job = client.jobs.get(job_id)

    if job.status == JobStatus.SUCCEEDED:
        print(job.result)
    elif job.status == JobStatus.FAILED:
        print(job.error)
    else:
        print(f"Still {job.status}")
```

| Status      | Meaning                           |
| ----------- | --------------------------------- |
| `queued`    | Accepted and waiting to start     |
| `running`   | Processing                        |
| `succeeded` | Result is ready                   |
| `failed`    | Processing ended with `job.error` |
| `cancelled` | The job was cancelled             |

Treat status values as open-ended. Use `job.is_terminal` or the exported
`JobStatus` members instead of assuming this table can never grow.

## Handle failures and timeouts

By default, `jobs.wait` raises `JobFailedError` for failed or cancelled jobs
and `JobTimeoutError` when its local wait budget expires. Both exceptions carry
the last job state.

```python theme={"dark"}
from ndi_sdk import JobFailedError, JobTimeoutError, NdiClient

with NdiClient() as client:
    try:
        job = client.jobs.wait(job_id, timeout=600)
    except JobFailedError as exc:
        print(exc.job.status, exc.job.error)
    except JobTimeoutError as exc:
        print(f"{exc.job.job_id} is still {exc.job.status}")
```

Set `raise_on_failure=False` when you prefer to inspect a failed or cancelled
job as a normal return value.

```python theme={"dark"}
job = client.jobs.wait(job_id, raise_on_failure=False)
if job.status != JobStatus.SUCCEEDED:
    print(job.error)
```

## Cancel a job

```python theme={"dark"}
cancelled = client.jobs.cancel(
    job_id,
    reason="The source document was replaced",
)
print(cancelled.status)
```

Only queued or running jobs can be cancelled. Cancelling a terminal job raises
`ConflictError` with code `job_not_cancellable`.

## List and paginate jobs

`jobs.list` returns one cursor page, newest first. Filter by workspace, kind,
status, or creation time.

```python theme={"dark"}
from datetime import UTC, datetime

from ndi_sdk import JobKind, JobStatus, NdiClient

with NdiClient() as client:
    page = client.jobs.list(
        kind=[JobKind.PARSE],
        status=[JobStatus.SUCCEEDED],
        created_after=datetime(2026, 9, 1, tzinfo=UTC),
        limit=50,
    )
    for job in page.items:
        print(job.job_id, job.created_at)
```

Use `iter_all` to traverse every page without managing cursors.

```python theme={"dark"}
with NdiClient() as client:
    for job in client.jobs.iter_all(workspace_id=workspace_id):
        print(job.job_id, job.status)
```

## Stream progress

`jobs.events` yields Server-Sent Events until the job becomes terminal or the
server closes the stream.

```python theme={"dark"}
with NdiClient() as client:
    for event in client.jobs.events(job_id):
        print(event.status, event.message)
```

Events are a latency convenience, not a completion guarantee. Always confirm
the final state with `jobs.get` or `jobs.wait`. Pass `last_event_id=` when
reconnecting to continue the event sequence.

## Inspect or delete a job

For document operations, `jobs.request` returns the request settings used by a
stored job:

```python theme={"dark"}
request = client.jobs.request(job_id)
print(request.request_type)
```

`jobs.delete(job_id)` soft-deletes a job from listings. Billing records remain,
and reading the job directly still returns its `deleted_at` value. Deleting a
queued or running job cancels it first.

```python theme={"dark"}
client.jobs.delete(job_id)
```

## Result retention

The job row can outlive its result payload. Check `job.result_state` before
using an older result. When retention has removed the payload, request and
artifact reads raise `ResultExpiredError`.

## Methods

```text theme={"dark"}
def get(job_id: UUID | str) -> Job
def list(*, workspace_id=None, kind=None, status=None, created_after=None, cursor=None, limit=None) -> Page[Job]
def iter_all(...) -> Iterator[Job]
def wait(job_id, *, timeout: float = 300, raise_on_failure: bool = True) -> Job
def cancel(job_id, *, reason: str | None = None) -> Job
def request(job_id) -> JobRequestEcho          # Python only
def delete(job_id) -> None                    # Python only
def events(job_id, *, last_event_id=None) -> Iterator[JobEvent]
def ground_crop(job_id, artifact_ref, *, byte_range=None) -> bytes
def usage(*, period_start: date, period_end: date, workspace_id=None, group_by="day") -> UsageResponse
```

## Next steps

* [Async client](/sdks/python/async)
* [Error handling](/sdks/python/error-handling)
* [Jobs and artifacts guide](/guides/jobs)