Background jobs¶

Most data operations in ineepy — uploads, updates, and binary-format queries — are processed by the server asynchronously as background jobs. Understanding how jobs work lets you build pipelines that submit multiple operations in parallel, integrate with task queues, or implement custom retry logic.

How background jobs work¶

Upload and update jobs — when you call submit_upload or submit_update, ineepy performs a two-phase handshake:

Allocate — ineepy POSTs to the API server. The server allocates a job, generates a presigned S3 upload URL, and responds with {job_id, upload_url}.
Transfer — ineepy PUTs the data directly to S3, bypassing the API server.
Process — the server ingests the data from S3 asynchronously. The job ID lets you track progress.

Query jobs — when you call submit_query or submit_cross_query, ineepy POSTs a query specification to POST /data/queries. No data is transferred:

Submit — ineepy POSTs the query spec. The server allocates a job and responds with {job_id, status}.
Process — the server runs the query and writes the result file to S3.
Download — poll_download polls GET /data/query/{job_id} until the server returns a 307 redirect to the presigned result file, then downloads the bytes.

The submit_* variants return the job ID immediately. The blocking variants (upload, update, cross_query, and query for parquet/csv) do the same thing internally but then poll until completion.

When to use `submit_*` vs blocking methods¶

Situation	Recommended method
Simple pipeline, one operation at a time	`upload` / `update`
JSON query, synchronous result	`query(format='json')`
Binary query result (parquet / CSV), single table	`query(format='parquet')` or `submit_query` + `poll_download`
Cross-table query with joins or nested filters	`cross_query` / `submit_cross_query` + `poll_download`
Submit multiple operations in parallel	`submit_upload` / `submit_update`
Decouple submission from completion tracking	`submit_upload` / `submit_update`
Integrate with an async task queue	`submit_upload` / `submit_update`
Custom timeout or retry logic per job	`submit_*` + `get_job`

Job lifecycle¶

Every job transitions through these states:

PENDING → PROCESSING → COMPLETED
                     ↘ FAILED

PENDING — job is queued, data transfer to S3 is complete
PROCESSING — server is ingesting or querying
COMPLETED — operation succeeded
FAILED — operation failed; see job.error for details

`DataJobResponse` fields¶

get_job() and all blocking data methods return a DataJobResponse:

Field	Type	Description
`job_id`	`str`	Unique job identifier
`table_id`	`str \\| None`	Target table
`job_type`	`DataJobType`	`UPLOAD`, `UPDATE`, `DOWNLOAD`, or `QUERY`
`status`	`DataJobStatus`	Current lifecycle state
`error`	`str \\| None`	Error message when `FAILED`
`update_options`	`UpdateOptions \\| None`	Mode and key columns for UPDATE jobs
`filters`	`list[Filter] \\| None`	Filters for DOWNLOAD jobs
`created_at`	`datetime`	When the job was created
`updated_at`	`datetime`	Last status change

Compare against the DataJobStatus enum from ineepy.helpers:

from ineepy.helpers import DataJobStatus

job = client.data.get_job(job_id)
if job.status == DataJobStatus.COMPLETED:
    ...

Submitting an upload¶

submit_upload transfers data to S3 and returns a job ID. You then poll get_job until the ingestion finishes:

        upload_data = json.dumps([
            {
                'user_id': 3,
                'event_type': 'signup',
                'recorded_at': '2024-01-02T10:00:00',
            },
        ]).encode()

        job_id = client.data.submit_upload(
            table_id='raw.jobs.events_b',
            data=upload_data,
            format='json',
        )
        print(f'Job submitted: {job_id}')

        while True:
            job = client.data.get_job(job_id)
            if job.status == DataJobStatus.COMPLETED:
                print(f'Completed at {job.updated_at}')
                break
            if job.status == DataJobStatus.FAILED:
                raise RuntimeError(f'Job failed: {job.error}')
            print(f'Status: {job.status.value}')
            time.sleep(2)

The polling loop above checks job.status.value to print a human-readable string (e.g. 'PENDING'). The comparison against DataJobStatus.COMPLETED uses the enum directly.

Submitting an update¶

submit_update works the same way and also exposes the update_options field once the job completes:

        more_events = json.dumps([
            {
                'user_id': 4,
                'event_type': 'logout',
                'recorded_at': '2024-01-02T18:00:00',
            },
        ]).encode()

        job_id = client.data.submit_update(
            table_id='raw.jobs.events',
            data=more_events,
            format='json',
            update_mode='append',
        )
        print(f'Update submitted: {job_id}')

        while True:
            job = client.data.get_job(job_id)
            if job.status == DataJobStatus.COMPLETED:
                print(
                    f'Update complete '
                    f'({job.update_options.mode})'
                )
                break
            if job.status == DataJobStatus.FAILED:
                raise RuntimeError(f'Job failed: {job.error}')
            time.sleep(2)

Running jobs concurrently¶

Because blocking methods wait for each job before returning, running N uploads sequentially takes N × server-processing-time. With submit_* you can submit all jobs upfront, then track them together:

        batch_a = json.dumps([
            {
                'user_id': 10,
                'event_type': 'login',
                'recorded_at': '2024-02-01T08:00:00',
            },
        ]).encode()
        batch_b = json.dumps([
            {
                'user_id': 20,
                'event_type': 'signup',
                'recorded_at': '2024-02-01T09:00:00',
            },
        ]).encode()

        job_id_a = client.data.submit_update(
            table_id='raw.jobs.events',
            data=batch_a,
            format='json',
            update_mode='append',
        )
        job_id_b = client.data.submit_update(
            table_id='raw.jobs.events_b',
            data=batch_b,
            format='json',
            update_mode='append',
        )
        print(f'Submitted: {job_id_a}, {job_id_b}')

        pending = {job_id_a, job_id_b}
        while pending:
            done = set()
            for jid in list(pending):
                job = client.data.get_job(jid)
                if job.status == DataJobStatus.COMPLETED:
                    print(f'Job {jid[:8]}… done')
                    done.add(jid)
                elif job.status == DataJobStatus.FAILED:
                    raise RuntimeError(
                        f'Job {jid} failed: {job.error}'
                    )
            pending -= done
            if pending:
                time.sleep(2)
        print('All jobs complete')

The polling loop above tracks a pending set of job IDs, removing each one as it reaches COMPLETED. This pattern scales to any number of concurrent jobs without blocking on each one individually.

Downloading query results¶

For parquet and CSV formats, submit_query POSTs a query spec to POST /data/queries and returns a job ID. poll_download then polls GET /data/query/{job_id} until the server issues a 307 redirect to the presigned result file, then downloads and returns the raw bytes:

        job_id = client.data.submit_query(
            table_id='raw.jobs.events',
            format='parquet',
        )
        print(f'Query submitted: {job_id}')

        result = client.data.poll_download(job_id)
        print(f'Downloaded {len(result)} bytes of parquet data')

Tip

query(format='json') is synchronous and returns a QueryResponse directly — no job ID involved. Use submit_query + poll_download (or the high-level query(format='parquet'|'csv')) only when you need binary output.

Error handling¶

Two exceptions are specific to jobs:

JobFailedError — raised when the server reports FAILED status. Attributes:

job_id — the failed job's ID
error — the server's error message

TimeoutError — raised when polling exceeds the configured timeout. Attribute:

job_id — the job that timed out

from ineepy.exceptions import JobFailedError, TimeoutError

try:
    job = client.data.upload(
        table_id='my.table',
        data=data,
        format='json',
    )
except JobFailedError as e:
    print(f'Job {e.job_id} failed: {e.error}')
except TimeoutError as e:
    print(f'Job {e.job_id} timed out')
    # Resubmit or increase timeout

When using submit_* and polling manually, raise these exceptions yourself:

if job.status == DataJobStatus.FAILED:
    raise JobFailedError(job.job_id, job.error)

Adjusting timeout and poll interval¶

The blocking methods accept timeout (seconds, default 300) and poll_interval (seconds, default 2):

job = client.data.upload(
    table_id='my.large.table',
    data=data,
    format='parquet',
    timeout=900,       # 15 minutes
    poll_interval=5,   # check every 5 seconds
)

When polling manually there is no built-in timeout — add one with time.monotonic():

import time

deadline = time.monotonic() + 900
while True:
    job = client.data.get_job(job_id)
    if job.status == DataJobStatus.COMPLETED:
        break
    if job.status == DataJobStatus.FAILED:
        raise RuntimeError(job.error)
    if time.monotonic() >= deadline:
        raise TimeoutError(job_id)
    time.sleep(5)

Async variant¶

All submit_* and get_job methods are available on AsyncIneepy with await. See Async background jobs for patterns using asyncio.gather to poll multiple jobs concurrently.