Data pipeline¶

This guide walks a table through its full life cycle: stream an initial load from disk, inspect and query it, keep a cached copy fresh with ETags, apply the three write modes, stream a large result back to disk, and clean up rows.

1. Set up the table¶

        ns, _ = client.namespaces.create(
            namespace_id='raw.events',
            title='Events',
        )
        schema = [
            ColumnSchema(name='user_id', data_type='int64'),
            ColumnSchema(name='event_type', data_type='string'),
            ColumnSchema(name='recorded_at', data_type='datetime'),
            ColumnSchema(name='value', data_type='float64'),
        ]
        table, _ = client.tables.create(
            table_id='raw.events.log',
            columns=schema,
            sort_by=[{'name': 'recorded_at'}],
        )

2. Initial load, streamed from disk¶

upload() performs the initial load of a table. data accepts raw bytes or an open file object — files are streamed to storage without buffering the whole payload in memory, so this works for files larger than RAM:

        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}')

Note

The server accepts upload() only while the table has never ingested data; every later write must use append, upsert, or overwrite. See Streaming large transfers for the accepted input types.

3. Inspect and query¶

info() returns snapshot statistics without scanning data:

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

JSON queries return rows and pagination metadata synchronously:

        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)

Filters use field:operator:value syntax (a single string or an iterable of them):

        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}')

4. Cache with ETags¶

Every JSON query carries an ETag tied to the table's current snapshot. Send it back as if_none_match — when nothing changed, the server answers 304 and query() returns None, so you skip the transfer and reuse your cached rows:

        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')

5. Apply updates¶

append() inserts rows without deduplication:

        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 matching all key_columns and 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() replaces the entire dataset:

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

6. Download large results to disk¶

For parquet or CSV output, submit the query as a job and stream the result straight to a file — poll_download_to() writes chunks as they arrive instead of holding the whole result in memory:

        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}')

Note

A failed download can leave a partial file at the destination — no atomic rename is performed.

7. Delete rows and truncate¶

delete_rows() removes rows matching every filter:

        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. It is processed asynchronously — poll info() before writing again, and reload with append() or overwrite() (the table is no longer eligible for upload()):

        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}')

8. Monitor jobs yourself¶

Every write has a submit_* variant returning a job ID immediately, which you poll with get_job():

        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 concurrent-job patterns.