cqlite-py

Python bindings for CQLite - a high-performance library for reading Apache Cassandra 5.0 SSTable files locally, without requiring a running Cassandra cluster.

Installation

pip install cqlite-py

Quick Start

import cqlite

# Open a database with schema
with cqlite.open('path/to/sstables', schema='schema.cql') as db:
    # Execute queries
    for row in db.execute('SELECT * FROM keyspace.table LIMIT 10'):
        print(row.to_dict())

Features

• Zero cluster dependency - Read SSTable files directly from disk
• Full CQL type support - All primitive types, collections, UDTs, and frozen types
• Memory-efficient streaming - Iterate over large datasets without loading all rows
• Thread-safe database handles - Safe concurrent access from multiple threads
• Cross-platform - Linux (x86_64, ARM64), macOS (Intel, Apple Silicon), Windows

Supported Platforms

Requirements

• Python 3.9+ — the boundaries are CI-tested, not merely advertised. CI installs the abi3 wheel on 3.9 (the floor), 3.14 (the current top of the range) and 3.13 (an additional compatibility leg), running an import plus one real query against the canonical corpus on each (Floor smoke in .github/workflows/python-ci.yml, issue #1459).

  That check is a binding matrix tier, so per repo CI-cost policy it runs on every push to main, on the nightly schedule, and on PRs labeled ci:bindings-full — not on a routine unlabeled PR. main is therefore continuously floor-tested and releases are cut from it, but a floor break is caught just after merge rather than before it.

  The range is unbounded above, so its top is the newest released CPython — when 3.15 ships it belongs in that matrix.

• Cassandra 5.0 SSTable files

API Reference

Opening a Database

import cqlite

# Context manager (recommended)
with cqlite.open(data_dir, schema=schema_path) as db:
    # use db...

# Manual management
db = cqlite.open(data_dir, schema=schema_path)
# use db...
db.close()

Cleanup on garbage collection (safety net, not the recommended path)

A handle that is garbage-collected without close() still cleans up best-effort: the write engine is closed (flushing any remaining memtable to an SSTable), the read engine's shutdown hook is called (today a no-op), and buffered telemetry is flushed. Some things are worth knowing about relying on it:

• It runs with the GIL held, because CPython frees the object from its deallocator. The flush and fsync therefore block other Python threads for their duration, where close() releases the GIL around the same work. Prefer with or an explicit close() in threaded code.
• It is best-effort by design. If the handle is dropped from inside a running Tokio runtime context — CQLite driven from a Rust async host — the cleanup is skipped rather than risk an unsafe teardown, so unflushed rows stay in the write-ahead log (replayable) instead of being flushed. close() is the only path with a guarantee.
• A Python asyncio event loop is NOT such a context. These bindings have no asyncio integration, so an asyncio thread has no Tokio runtime and nothing is skipped: a handle collected there runs the full flush + fsync with the GIL held, blocking the event loop (and up to ~5s more if OpenTelemetry export is enabled and the collector is unreachable). In asyncio code, close handles explicitly — ideally off the loop thread.

A StreamingIterator that outlives its Database is unaffected by the handle being collected, for both read-only and writable handles: it keeps yielding its remaining rows. Its rows come from a background task holding its own reference to the storage engine, so dropping the handle cannot stop the stream, and this cleanup deliberately does not invalidate it. (An explicit close() still does — that is a user stating intent, and is unchanged; see issue #1462.)

Executing Queries

# Simple query
results = db.execute('SELECT * FROM keyspace.table')
for row in results:
    print(row.to_dict())

# With LIMIT
for row in db.execute('SELECT name, age FROM users LIMIT 100'):
    print(f"{row['name']}: {row['age']}")

# Access query metadata
print(f"Rows returned: {len(results)}")
print(f"Execution time: {results.execution_time_ms}ms")
print(f"Columns: {[col.name for col in results.columns]}")

Streaming Large Results

For memory-efficient iteration over large datasets:

from cqlite import StreamingConfig

# Configure streaming for memory efficiency
config = StreamingConfig(buffer_size=512, chunk_size=1000)
for row in db.execute_streaming('SELECT * FROM large_table', config=config):
    process(row)

# Track progress
iterator = db.execute_streaming('SELECT * FROM large_table')
for row in iterator:
    if iterator.rows_received % 10000 == 0:
        print(f"Processed {iterator.rows_received} rows")

Configuration Presets

import cqlite

# Built-in presets for common use cases
config = cqlite.memory_optimized()      # 256 MB max memory
config = cqlite.performance_optimized() # 4 GB max memory

# Open database with preset configuration
db = cqlite.open('path/to/data', schema='schema.cql', config='memory_optimized')

# Validate custom configuration
custom_config = {'memory': {'max_memory': 536870912}}  # 512 MB
cqlite.validate_config(custom_config)

Refreshing SSTables (v0.13)

If Cassandra (or another process) writes new SSTables while your database handle is open, call refresh() to re-discover them. Refresh is explicit-only (CQLite never rescans behind your back) and atomic / fail-closed: if any newly found generation fails to open, the swap is rolled back and the handle keeps serving the prior, consistent set of readers.

import cqlite

with cqlite.open('path/to/sstables', schema='schema.cql') as db:
    # ... time passes; Cassandra flushes/compacts new SSTables to disk ...

    report = db.refresh()
    print(f'Tables scanned:  {report.tables_scanned}')
    print(f'Readers added:   {report.readers_added}')
    print(f'Readers removed: {report.readers_removed}')

    # Subsequent queries see the newly discovered data
    for row in db.execute('SELECT * FROM keyspace.table'):
        print(row.to_dict())

refresh() returns a RefreshReport with the integer attributes tables_scanned, readers_added, and readers_removed, plus a to_dict() helper. It raises RuntimeError if the database is already closed, and CqliteError if a newly discovered generation fails to open (the prior reader set is preserved).

Result Byte Budget (v0.13)

Non-streaming execute() queries are bounded by a result-size budget, defaulting to 64 MiB (64 * 1024 * 1024 bytes). When the materialized result's running byte estimate exceeds the budget, the query fails with a cqlite.QueryError directing you to add a LIMIT clause or switch to execute_streaming(). Streaming queries are not subject to this budget.

Adjust the budget with the max_result_bytes key in the config passed to cqlite.open() (absent, it stays at 64 MiB):

import cqlite

# Raise the budget to 256 MiB for this handle
config = {'max_result_bytes': 256 * 1024 * 1024}

try:
    with cqlite.open('path/to/data', schema='schema.cql', config=config) as db:
        rows = db.execute('SELECT * FROM keyspace.big_table')
        for row in rows:
            process(row)
except cqlite.QueryError as e:
    # Result exceeded the byte budget — add a LIMIT or stream instead
    print(f'Result too large: {e}')
    for row in db.execute_streaming('SELECT * FROM keyspace.big_table'):
        process(row)

OpenTelemetry Tracing (v0.13)

CQLite can emit OpenTelemetry traces when built with the observability Cargo feature; without that feature the configuration is accepted but is a no-op. Pass an otel_config dict to cqlite.open(). Values are layered over the CQLITE_OTEL_* environment variables, and OpenTelemetry is initialized once per process.

import cqlite

db = cqlite.open(
    'path/to/sstables',
    schema='schema.cql',
    otel_config={
        'enabled': True,                       # default False
        'endpoint': 'http://localhost:4317',   # default 'http://localhost:4317'
        'protocol': 'grpc',                    # 'grpc' (default) or 'http'
        'service_name': 'cqlite',              # default 'cqlite'
        'service_version': '0.13.0',           # default: package version
        'sampling_ratio': 1.0,                 # default 1.0
        'timeout_ms': 10000,                   # default 10000
    },
)

Unknown keys raise ValueError.

Error Handling

import cqlite

try:
    with cqlite.open('path/to/data', schema='schema.cql') as db:
        result = db.execute('SELECT * FROM keyspace.table')
        for row in result:
            print(row.to_dict())
except cqlite.ParseError as e:
    print(f"Query syntax error: {e}")
except cqlite.QueryError as e:
    print(f"Query execution failed: {e}")
except cqlite.SchemaError as e:
    print(f"Schema validation failed: {e}")
except IOError as e:
    print(f"File not found: {e}")
except RuntimeError as e:
    print(f"Database already closed: {e}")

Exception Hierarchy:

CqliteError (base exception)
├── SchemaError   - Schema parsing or validation failures
├── QueryError    - Query execution failures
└── ParseError    - CQL syntax errors

Built-in exceptions also used:
├── IOError       - File system errors
├── ValueError    - Invalid configuration
├── RuntimeError  - Invalid state (e.g., database closed)
└── MemoryError   - Memory allocation failures

Type Conversions

CQL types are automatically converted to Python native types:

UDT type identity is carried out of band

A CQL user-defined type decodes to a cqlite.Udt:

udt = row["address"]
udt.type_name          # 'address_type'  — the declared UDT type
udt.keyspace           # 'test_collections'
udt.fields             # mappingproxy({'street': '1 Main St', 'city': 'SF'}) — declared fields ONLY
udt["street"]          # mapping access, delegating to .fields
"city" in udt, len(udt), sorted(udt.keys())

Breaking change (issue #3504). _type and _keyspace used to be injected as dict keys, i.e. into the same namespace as the UDT's own field names — so a UDT declaring a field named _type or _keyspace (legal CQL via a quoted identifier) silently overwrote the marker and the type name became unrecoverable. Migration:

udt["_type"] now reaches a FIELD of that name, raising KeyError when the UDT declares none. Udt is frozen, and equality/hashing are over (keyspace, type_name, fields), so it can be used as a dict key whenever its field values are hashable. udt.fields is therefore a read-only types.MappingProxyType view: udt.fields["z"] = 1 raises TypeError rather than moving a Udt already used as a key out of its hash bucket. Take dict(udt.fields) for a mutable copy.

CQL decimal rendering policy

A CQL decimal is unscaled x 10^(-scale) where unscaled is an arbitrary-precision two's-complement integer. Both CQLite language bindings share one implementation and one policy (issue #1452), so a value can never render in one binding and be refused by the other:

A negative scale multiplies by a power of ten, so there is no positional form for it and none of the size thresholds apply. A consumer that parses these strings must therefore accept exponent form at any magnitude: ^-?[0-9]+(\.[0-9]+)?$ alone is not sufficient.

Below the 32 KiB ceiling the render is infallible: a well-formed value always renders, whatever its scale. Above it the refusal is a typed, catchable error — a corrupt SSTable never aborts the host process.

Write Operations

CQLite v0.9.0 adds write support to the Python bindings. Open the database with writable=True and a write_dir to enable write operations.

import cqlite

with cqlite.open(
    'path/to/sstables',
    schema='schema.cql',
    writable=True,
    write_dir='/tmp/my-writes',
) as db:
    # Write rows via CQL INSERT, UPDATE, or DELETE
    db.execute(
        "INSERT INTO test_basic.simple_table (id, name, age) "
        "VALUES (11111111-1111-1111-1111-111111111111, 'Alice', 30)"
    )
    db.execute(
        "UPDATE test_basic.simple_table SET age = 31 "
        "WHERE id = 11111111-1111-1111-1111-111111111111"
    )

    # Flush the in-memory write buffer (memtable) to an SSTable on disk.
    # Returns the path to the flushed Data.db file.
    path = db.flush_run()
    print(f'Flushed to: {path}')

    # Run background compaction within a time budget
    report = db.maintenance_step(budget_ms=100)
    print(f'Merged {report.rows_merged} rows in {report.time_spent_ms:.1f} ms')
    if report.pending_compaction:
        print('More compaction work available')

    # Inspect write statistics
    stats = db.write_stats
    print(f'Memtable size: {stats.memtable_size_bytes} bytes')
    print(f'Total flushed: {stats.total_written_bytes} bytes')

Write API

Known Limitations

• Counter columns cannot be written — execute() raises CqliteError for counter mutations.
• Concurrent queries on the same handle may need a warm-up query first (Issue #311).

See docs/write-support-limitations.md for the full limitations reference.

Resources

• Acceptance Testing Notebook - Interactive examples and validation
• Type Stubs - Complete API type hints for IDE support
• Main Project README - CQLite project overview and documentation
• Write Support Guide - Detailed write documentation
• Issue Tracker - Report bugs or request features

License

MIT OR Apache-2.0

Links

• GitHub Repository
• PyPI Package
• Documentation