pystreamci 0.3.1

Python client library for StreamCI

PyStreamCI

Python client library for StreamCI — a connector-style API in the spirit of database drivers.

Installation

pip install pystreamci
# or editable install for development
pip install -e clients/python/

Quick Start

import pystreamci

# Secret-key auth (recommended) — use as context manager
with pystreamci.connect("https://api.streamci.org",
                        target="my_sensors",
                        secret_key="...") as client:

    # Insert a document
    client.insert({"time": {"$date": "2025-08-04T10:00:00Z"}, "val": 42})

    # Insert many documents (auto-batched)
    client.insert_many(records, batch_size=1000)

    # Upload an NDJSON/CSV file
    client.insert_file("data.ndjson")

    # Query (returns list of dicts)
    docs = client.query({"val": {"$gte": 10}}, sort={"time": 1}, limit=50)

    # Stream large results with an iterator
    for doc in client.query_iter({"val": {"$gte": 0}}):
        process(doc)

    # Download a blob field
    blob = client.query_blob(document_id="67abc...", field="image1")

    # Download blob fields from multiple documents
    blobs = client.query_blobs_by_field(["67abc...", "89def..."], field="image1")

    # Download all blob fields from a single document
    all_blobs = client.query_blobs_by_doc(document_id="67abc...")

    # Insert with blob attachment
    with open("photo.jpg", "rb") as f:
        client.insert({"name": "doc"}, blobs={"photo": ("photo.jpg", f, "image/jpeg")})

    # Update and delete
    client.update({"val": 42}, {"val": 43})
    client.delete({"val": 43})

# Password auth
client = pystreamci.connect("https://api.streamci.org",
                            target="my_sensors",
                            username="user1", password="pass123")

API Reference

pystreamci.connect(host, target, *, secret_key=None, username=None, password=None, max_retries=3, timeout=30.0, log_level="WARNING", log_file=None)

Returns a StreamCIClient. Auth type is inferred from the provided credentials.

StreamCIClient

Method	Description
insert(data, blobs=None)	Insert a single document (with optional blob attachments)
insert_many(data, datatype, batch_size)	Batch insert (auto-splits at batch_size)
insert_file(file_path, datatype)	Upload NDJSON/CSV file
update(filter, data, blobs=None)	Update matching documents
delete(filter)	Delete matching documents
query(query, *, project, sort, limit)	Query and return list
query_iter(query, *, project, sort, limit)	Query with streaming iterator
query_blob(document_id, field)	Download blob field as bytes
query_blobs_by_field(document_ids, field)	Download blob field from multiple documents
query_blobs_by_doc(document_id, *, ignore_errors)	Download all blob fields from a document (including nested blob fields); returns dot-notation keys

pystreamci.make_date(dt)

Convert a datetime or ISO-8601 string to a StreamCI {"$date": "..."} object.

pystreamci.make_date("2025-08-04T10:00:00Z")
pystreamci.make_date(datetime(2025, 8, 4, 10, 0, tzinfo=timezone.utc))

Exceptions

Exception	When raised
StreamCIConnectionError	Connection failed after retries
StreamCIAuthError	Authentication/authorization error
StreamCIRequestError	Server returned status=0
StreamCIParseError	Response parsing failed
StreamCITimeoutError	Request timed out
StreamCIValidationError	Invalid client-side input

query() vs query_iter()

Method	Returns	Use when
query(...)	list[Document]	Result fits in memory; need random access
query_iter(...)	Iterator[Document]	Large result sets; process one doc at a time

query_iter() uses HTTP streaming (stream=True) and parses NDJSON line-by-line, keeping memory usage constant regardless of result size.