Data¶

The data resource manages all table data operations. Every write (upload, update) is processed by a background job on the server — ineepy handles the polling for you by default, or you can manage job IDs yourself for more control.

Methods¶

Dataset info¶

info() returns snapshot statistics without scanning the full table.

        stats = client.data.info(table_id='raw.events.log')
        print(f'Rows: {stats.row_count}')
        print(f'Size: {stats.total_size_bytes} bytes')

Fields available on DatasetInfoResponse: row_count, total_size_bytes, data_files, last_updated, snapshot_id.

Uploading data¶

upload() and submit_upload() perform the initial load of a table: the server rejects them with a ConflictError (409) once the table has ever ingested data — even after a truncate — so every subsequent write goes through append, upsert, or overwrite.

        with open('events.csv', 'rb') as f:
            job = client.data.upload(
                table_id='raw.events.log',
                data=f,
                format='csv',
            )
        print(f'Streamed upload from disk: {job.status}')

        job_id = client.data.submit_upload(
            table_id='raw.events.log',
            data=json.dumps(extra_events).encode(),
            format='json',
        )
        print(f'Upload submitted: {job_id}')

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

See Background jobs for patterns around managing job IDs.

Querying data¶

        result = client.data.query(
            table_id='raw.events.log',
            format='json',
        )
        print(f'Total events: {result.metadata.total_count}')
        for row in result.items:
            print(row)

result = client.data.query(table_id='raw.events.log', format='json', limit=100, offset=0)
while result.metadata.has_next:
    offset += 100
    result = client.data.query(table_id='raw.events.log', format='json', limit=100, offset=offset)

        job_id = client.data.submit_query(
            table_id='raw.events.log',
            format='parquet',
        )
        print(f'Query submitted: {job_id}')
        result = client.data.poll_download(job_id)
        print(f'Downloaded {len(result)} bytes')

Conditional queries (caching)¶

Every format='json' response includes an ETag tied to the table's current snapshot. Pass it back as if_none_match on subsequent calls — the server returns 304 Not Modified and query() returns None when the data has not changed, letting you skip re-processing and avoid unnecessary transfer:

        result = client.data.query(
            table_id='raw.events.log',
            format='json',
        )
        etag = result.etag

        cached = client.data.query(
            table_id='raw.events.log',
            format='json',
            if_none_match=etag,
        )
        if cached is None:
            print('Data unchanged — using the cached copy')

When the data has changed, the response is a fresh QueryResponse — read result.etag again to update your cached ETag.

etag is '' when the table has no snapshot yet (no data has been ingested). if_none_match is ignored for format='parquet' and format='csv'.

Filtering results¶

Filters use field:operator:value format and are combined with AND semantics:

        login_events = client.data.query(
            table_id='raw.events.log',
            filters=['event_type:eq:login'],
            format='json',
        )
        print(f'Login events: {login_events.metadata.total_count}')

Common operators: eq, ne, lt, lte, gt, gte.

Cross-table queries¶

cross_query() and submit_cross_query() use the POST /data/queries endpoint to run async queries across multiple tables with joins and nested AND/OR filter predicates. Results are always returned as a file (parquet, csv, or json) via a presigned S3 URL.

from ineepy import CrossQueryRequest, FilterGroup, FilterLeaf, JoinSpec

req = CrossQueryRequest(
    table_id='raw.sales.orders',
    filters=FilterGroup(
        type='and',
        predicates=[
            FilterLeaf(field='amount', operator='gt', value=50),
            FilterLeaf(field='status', operator='eq', value='active'),
        ],
    ),
    joins=[
        JoinSpec(
            table_id='raw.customers',
            how='inner',
            left_on='customer_id',
            right_on='id',
        )
    ],
    select=['order_id', 'name', 'amount'],
    format='parquet',
)

# Blocking — waits until the result is ready and returns bytes
parquet_bytes = client.data.cross_query(req)

# Non-blocking — returns the job ID immediately
job_id = client.data.submit_cross_query(req)
parquet_bytes = client.data.poll_download(job_id)

FilterLeaf holds a single condition. FilterGroup combines two or more predicates with type='and' or type='or' and requires at least two predicates. Use FilterLeaf directly for a single condition.

Supported JoinSpec.how values: inner, left, right, full, cross, semi, anti.

Updating data¶

Three explicit write methods cover the most common modes.

Append¶

Inserts new rows without deduplication — existing rows are untouched:

        job = client.data.append(
            table_id='raw.events.log',
            data=json.dumps(new_events).encode(),
            format='json',
        )
        print(f'Appended: {job.status}')

Upsert¶

Updates rows that match all key_columns; inserts the rest:

        job = client.data.upsert(
            table_id='raw.events.log',
            data=json.dumps(upsert_data).encode(),
            format='json',
            key_columns=['user_id', 'event_type'],
        )
        print(f'Upserted: {job.status}')

Overwrite¶

Deletes all existing rows and replaces them with the new dataset:

        job = client.data.overwrite(
            table_id='raw.events.log',
            data=json.dumps(fresh_data).encode(),
            format='json',
        )
        print(f'Overwritten: {job.status}')

Non-blocking writes¶

Every write method has a submit_* counterpart that returns a job ID immediately instead of polling. Use this when you want to submit multiple operations in parallel or integrate with your own job-tracking logic:

        job_id = client.data.submit_append(
            table_id='raw.events.log',
            data=json.dumps(extra_update).encode(),
            format='json',
        )
        print(f'Update submitted: {job_id}')

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

Generic update() / submit_update()¶

The generic method is available when the mode is determined at runtime:

client.data.update(
    table_id='raw.events.log',
    data=data,
    format='csv',
    update_mode=mode,         # 'append', 'overwrite', or 'upsert'
    key_columns=keys,         # required when mode == 'upsert'
)

Deleting rows¶

Deletes all rows matching every filter (AND semantics). At least one filter is required:

        client.data.delete_rows(
            table_id='raw.events.log',
            filters=['event_type:eq:logout'],
        )
        print('Deleted logout events')

Truncate¶

Removes all rows while preserving the schema. Truncation is processed asynchronously — poll info() until the table reports empty before writing again. A truncated table is not eligible for upload() again; reload it with append() or overwrite():

        client.data.truncate(table_id='raw.events.log')

        # Truncation is processed asynchronously — wait until it lands
        while client.data.info(table_id='raw.events.log').row_count:
            time.sleep(1)
        print('Table truncated')

        job = client.data.append(
            table_id='raw.events.log',
            data=json.dumps(fresh_data).encode(),
            format='json',
        )
        print(f'Reloaded: {job.status}')

Streaming large transfers¶

Uploads and downloads can stream data instead of buffering entire payloads in memory, so transfers are bounded by disk, not RAM.

Uploads¶

Every write method's data parameter accepts, on the sync client:

• bytes — simplest, for payloads that already fit in memory.

• A file-like object (e.g. an open(..., 'rb') handle) — the file is streamed to storage; its size is announced via Content-Length, which every S3-compatible endpoint accepts. Prefer this for large files:

          with open('events.csv', 'rb') as f:
              job = client.data.upload(
                  table_id='raw.events.log',
                  data=f,
                  format='csv',
              )
          print(f'Streamed upload from disk: {job.status}')
  

• An iterable of bytes chunks — sent with chunked transfer encoding, because the total size is unknown up front. Some S3-compatible endpoints reject chunked uploads (HTTP 411 Length Required) — prefer a file-like object when the content lives on disk.

The async client accepts bytes or an async-iterable of bytes (the same chunked-transfer caveat applies to async iterables). Note the asymmetry: sync accepts file-likes and plain iterables; async accepts async-iterables.

Downloads¶

poll_download() buffers the whole result in memory and returns bytes. For large results, poll_download_to() streams the file straight to disk and returns the written Path:

        job_id = client.data.submit_query(
            table_id='raw.events.log',
            format='parquet',
        )
        path = client.data.poll_download_to(job_id, 'events.parquet')
        print(f'Saved {path.stat().st_size} bytes to {path}')

Supported formats¶

JSON queries are returned synchronously from GET /tables/{id}/data/query. Parquet and CSV queries submit an async job via POST /data/queries, then poll GET /data/query/{job_id} until the server issues a 307 redirect to the presigned result file.

Update modes¶