Python bindings guide

tsink ships Python bindings via UniFFI. The tsink package gives you the full storage engine — writes, queries, aggregation, rollups, snapshots — from Python with no server process required. After installation, import it as tsink. UniFFI record types such as Label, Row, and MetricSeries use keyword-only constructors in Python.

Installation

pip install tsink

Requires Python 3.8+. The wheel includes the native Rust library; no Rust toolchain needed at runtime.

Building from source

If a pre-built wheel is not available for your platform:

pip install maturin
cd crates/tsink-uniffi
maturin develop --release

Quick start

from tsink import TsinkStorageBuilder, DataPoint, Label, Row, Value

builder = TsinkStorageBuilder()
builder.with_data_path("./tsink-data")
db = builder.build()

db.insert_rows([
    Row(
        metric="cpu_usage",
        labels=[Label(name="host", value="web-1")],
        data_point=DataPoint(timestamp=1_700_000_000_000, value=Value.F64(v=42.0)),
    ),
])

points = db.select(
    "cpu_usage",
    [Label(name="host", value="web-1")],
    0,
    2_000_000_000_000,
)
for p in points:
    print(f"ts={p.timestamp} value={p.value}")

db.close()

TsinkStorageBuilder

Create a builder, call configuration methods, then call build() once to get a TsinkDB handle. The builder is consumed by build() and cannot be reused.

from datetime import timedelta
from tsink import (
    TsinkStorageBuilder,
    TimestampPrecision,
    WalSyncMode,
    WalReplayMode,
    StorageRuntimeMode,
)

builder = TsinkStorageBuilder()
builder.with_data_path("/var/lib/tsink")
builder.with_retention(timedelta(days=30))
builder.with_timestamp_precision(TimestampPrecision.MILLISECONDS)
builder.with_memory_limit(512 * 1024 * 1024)       # 512 MiB
builder.with_cardinality_limit(1_000_000)
builder.with_wal_sync_mode(WalSyncMode.PERIODIC(interval=timedelta(seconds=1)))
db = builder.build()

Configuration methods

All methods return None and mutate the builder in place.

Data & persistence

Method	Default	Description
`with_data_path(path)`	none	Directory for WAL, segments, and metadata.
`with_object_store_path(path)`	none	Path (or object-store prefix) for warm/cold tier segments.

Retention

Method	Default	Description
`with_retention(duration)`	14 days	Global retention window.
`with_retention_enforced(bool)`	`False`	Reject writes outside the retention window.
`with_tiered_retention_policy(hot, warm)`	none	Separate retention for hot and warm tiers.

Timestamp precision

Method	Default	Description
`with_timestamp_precision(precision)`	`Nanoseconds`	One of `Nanoseconds`, `Microseconds`, `Milliseconds`, `Seconds`.

Chunks & partitions

Method	Default	Description
`with_chunk_points(n)`	2048	Points per in-memory chunk before sealing.
`with_partition_duration(duration)`	1 hour	Time range per partition.
`with_max_active_partition_heads_per_series(n)`	8	Active partition heads per series (out-of-order fanout).

Memory & cardinality

Method	Default	Description
`with_memory_limit(bytes)`	unlimited	Memory budget; exceeding triggers backpressure.
`with_cardinality_limit(series)`	unlimited	Maximum unique series count.

Concurrency

Method	Default	Description
`with_max_writers(n)`	CPU count	Parallel writer threads.
`with_write_timeout(duration)`	30 s	Maximum wait for a writer slot.

WAL

Method	Default	Description
`with_wal_enabled(bool)`	`True`	Enable/disable the write-ahead log.
`with_wal_size_limit(bytes)`	unlimited	Maximum WAL size on disk.
`with_wal_buffer_size(size)`	default	In-memory WAL buffer size.
`with_wal_sync_mode(mode)`	`PerAppend`	`WalSyncMode.PER_APPEND` (crash-safe) or `WalSyncMode.PERIODIC(interval)`.
`with_wal_replay_mode(mode)`	`Strict`	`WalReplayMode.STRICT` or `WalReplayMode.SALVAGE`.

Remote segments

Method	Default	Description
`with_remote_segment_cache_policy(policy)`	`MetadataOnly`	Caching strategy for remote tier segments.
`with_remote_segment_refresh_interval(duration)`	default	Refresh interval for the remote segment catalog.
`with_mirror_hot_segments_to_object_store(bool)`	`False`	Mirror hot segments to the object store.

Runtime

Method	Default	Description
`with_runtime_mode(mode)`	`ReadWrite`	`StorageRuntimeMode.READ_WRITE` or `COMPUTE_ONLY`.
`with_background_fail_fast(bool)`	`True`	Halt on unrecoverable background errors.
`with_metadata_shard_count(n)`	auto	Number of metadata shards.

TsinkDB

TsinkDB is the main database handle returned by builder.build(). It is thread-safe and can be shared across Python threads.

Writing data

Value types

Values are represented by the Value enum:

Value.F64(v=3.14)              # 64-bit float
Value.I64(v=-42)               # 64-bit signed integer
Value.U64(v=100)               # 64-bit unsigned integer
Value.BOOL(v=True)             # boolean
Value.BYTES(v=b"\xCA\xFE")    # raw bytes
Value.STR(v="hello")           # UTF-8 string
Value.HISTOGRAM(v=histogram)   # native Prometheus histogram

Insert rows

rows = [
    Row(
        metric="http_requests_total",
        labels=[
            Label(name="method", value="GET"),
            Label(name="status", value="200"),
        ],
        data_point=DataPoint(timestamp=1_700_000_000_000, value=Value.F64(v=1027.0)),
    ),
    Row(
        metric="memory_free_bytes",
        labels=[Label(name="host", value="web-1")],
        data_point=DataPoint(timestamp=1_700_000_000_000, value=Value.I64(v=8_589_934_592)),
    ),
]

db.insert_rows(rows)

Write acknowledgement

insert_rows_with_result returns a WriteResult so you can inspect the durability guarantee:

result = db.insert_rows_with_result(rows)
print(result.acknowledgement)
# WriteAcknowledgement.DURABLE   — fsync'd (PerAppend WAL mode)
# WriteAcknowledgement.APPENDED  — in WAL buffer (Periodic mode)
# WriteAcknowledgement.VOLATILE  — in memory only (WAL disabled)

Querying data

Simple select

points = db.select(
    "cpu_usage",
    [Label(name="host", value="web-1")],
    start=0,
    end=2_000_000_000_000,
)

for p in points:
    print(p.timestamp, p.value)

Pass an empty label list to match the bare metric (no labels):

points = db.select("cpu_usage", [], start=0, end=2_000_000_000_000)

Select all label combinations

all_series = db.select_all("http_requests_total", start=0, end=2_000_000_000_000)

for labeled in all_series:
    tag_str = ", ".join(f"{l.name}={l.value}" for l in labeled.labels)
    print(f"[{tag_str}]: {len(labeled.data_points)} points")

Returns a list of LabeledDataPoints, each carrying labels and data_points.

Select multiple series at once

from tsink import MetricSeries

series = [
    MetricSeries(name="cpu_usage", labels=[Label(name="host", value="web-1")]),
    MetricSeries(name="cpu_usage", labels=[Label(name="host", value="web-2")]),
]

results = db.select_many(series, start=0, end=2_000_000_000_000)
for sp in results:
    print(f"{sp.series.name} {sp.series.labels}: {len(sp.points)} points")

Advanced queries with QueryOptions

from tsink import Aggregation, DownsampleOptions, QueryOptions

options = QueryOptions(
    labels=[Label(name="host", value="web-1")],
    start=0,
    end=2_000_000_000_000,
    aggregation=Aggregation.AVG,
    downsample=DownsampleOptions(interval=60_000),  # 1-minute buckets
    limit=1000,
    offset=0,
)

points = db.select_with_options("cpu_usage", options)

Available aggregations:

`Aggregation` variant	Description
`NONE`	No aggregation (default)
`SUM`	Sum of values
`MIN` / `MAX`	Minimum / maximum
`AVG`	Arithmetic mean
`FIRST` / `LAST`	First / last point in window
`COUNT`	Number of points
`MEDIAN`	Median value
`RANGE`	Max − Min
`VARIANCE` / `STD_DEV`	Statistical variance / standard deviation

Paginated row scanning

For large result sets, scan in pages:

from tsink import QueryRowsScanOptions

offset = None
while True:
    page = db.scan_metric_rows(
        "cpu_usage",
        start=0,
        end=2_000_000_000_000,
        options=QueryRowsScanOptions(max_rows=10_000, row_offset=offset),
    )
    process(page.rows)

    if page.truncated:
        offset = page.next_row_offset
    else:
        break

scan_series_rows works the same way but accepts a list of MetricSeries instead of a metric name.

Series discovery

List all metrics

all_metrics = db.list_metrics()
for m in all_metrics:
    print(m.name, m.labels)

list_metrics_with_wal() includes series that exist only in the WAL (not yet flushed).

Filter with matchers

from tsink import SeriesMatcher, SeriesMatcherOp, SeriesSelection

selection = SeriesSelection(
    metric="http_requests_total",
    matchers=[
        SeriesMatcher(name="method", op=SeriesMatcherOp.EQUAL, value="GET"),
        SeriesMatcher(name="status", op=SeriesMatcherOp.REGEX_MATCH, value="2.."),
    ],
    start=None,
    end=None,
)

matched = db.select_series(selection)

`SeriesMatcherOp`	Equivalent	Example
`EQUAL`	`=`	`method="GET"`
`NOT_EQUAL`	`!=`	`status!="500"`
`REGEX_MATCH`	`=~`	`host=~"web-.*"`
`REGEX_NO_MATCH`	`!~`	`env!~"staging\|dev"`

Deleting series

result = db.delete_series(
    SeriesSelection(metric="old_metric", matchers=[], start=None, end=None)
)
print(f"matched={result.matched_series}, tombstones={result.tombstones_applied}")

Deletion is tombstone-based. Tombstones are merged during compaction.

Rollup policies

Define materialized downsampled views:

from tsink import RollupPolicy

policies = [
    RollupPolicy(
        id="cpu_5m_avg",
        metric="cpu_usage",
        match_labels=[],
        interval=300_000,             # 5 minutes in ms
        aggregation=Aggregation.AVG,
        bucket_origin=0,
    ),
]

snapshot = db.apply_rollup_policies(policies)
for status in snapshot.policies:
    print(f"{status.policy.id}: materialized through {status.materialized_through}")

Trigger an immediate rollup run:

snapshot = db.trigger_rollup_run()

Snapshots

db.snapshot("/backups/tsink-2026-03-12")

# Restore to a new data directory.
TsinkStorageBuilder.restore_from_snapshot("/backups/tsink-2026-03-12", "/var/lib/tsink")

Observability

Inspect engine internals at runtime:

snap = db.observability_snapshot()

print(f"memory: {snap.memory.active_and_sealed_bytes} / {snap.memory.budgeted_bytes} bytes")
print(f"WAL: {snap.wal.segment_count} segments, {snap.wal.size_bytes} bytes")
print(f"compaction: {snap.compaction.runs_total} runs, {snap.compaction.errors_total} errors")
print(f"queries: {snap.query.select_calls_total} selects")

if snap.health.degraded:
    print(f"engine degraded: {snap.health.last_background_error}")

The snapshot covers memory, WAL, retention, flush pipeline, compaction, queries, rollups, remote storage, and overall health.

Memory inspection

print(f"used:   {db.memory_used()} bytes")
print(f"budget: {db.memory_budget()} bytes")

Closing the database

db.close()

Flushes remaining data, syncs the WAL, and shuts down background workers. Any method called after close() raises TsinkUniFFIError.

Native histograms

Store Prometheus-style native histograms:

from tsink import (
    NativeHistogram,
    HistogramBucketSpan,
    HistogramCount,
    HistogramResetHint,
)

histogram = NativeHistogram(
    count=HistogramCount.INT(v=10),
    sum=55.0,
    schema=3,
    zero_threshold=1e-128,
    zero_count=HistogramCount.INT(v=1),
    negative_spans=[],
    negative_deltas=[],
    negative_counts=[],
    positive_spans=[HistogramBucketSpan(offset=0, length=3)],
    positive_deltas=[2, 1, -1],
    positive_counts=[],
    reset_hint=HistogramResetHint.NO,
    custom_values=[],
)

db.insert_rows([
    Row(
        metric="request_duration",
        labels=[],
        data_point=DataPoint(
            timestamp=1_700_000_000_000,
            value=Value.HISTOGRAM(v=histogram),
        ),
    ),
])

Shard window operations

These methods support distributed deployments where data is partitioned across nodes. They are used internally by the cluster layer but are available for advanced use cases.

from tsink import MetadataShardScope, ShardWindowScanOptions

# Compute a digest (fingerprint) for a shard window.
digest = db.compute_shard_window_digest(
    shard=0, shard_count=16,
    window_start=0, window_end=2_000_000_000_000,
)
print(f"series={digest.series_count} points={digest.point_count} fp={digest.fingerprint}")

# Scan rows for a shard window with pagination.
page = db.scan_shard_window_rows(
    shard=0, shard_count=16,
    window_start=0, window_end=2_000_000_000_000,
    options=ShardWindowScanOptions(max_series=100, max_rows=10_000, row_offset=None),
)

# List metrics limited to specific shards.
scope = MetadataShardScope(shard_count=16, shards=[0, 1, 2])
metrics = db.list_metrics_in_shards(scope)

Error handling

All methods raise TsinkUniFFIError on failure. The exception carries a string message and is one of these variants:

Variant	When it occurs
`NoDataPoints`	No data found for the given metric/time range.
`InvalidTimeRange`	`start > end` or otherwise invalid range.
`StorageClosed`	Operation attempted after `close()`.
`InvalidInput`	Bad metric name, label, or unsupported operation.
`IoError`	File-system or disk I/O failure.
`DataCorruption`	Checksum mismatch or parse error.
`ResourceExhausted`	Memory budget, cardinality limit, or WAL size limit exceeded.
`Other`	Lock poisoning, channel errors, WAL issues, etc.

from tsink import TsinkUniFFIError

try:
    db.insert_rows(rows)
except TsinkUniFFIError as e:
    print(f"write failed: {e}")

Type reference

Records (dataclasses)

Python type	Fields	Notes
`Label`	`name: str`, `value: str`	Metric label key-value pair.
`DataPoint`	`timestamp: int`, `value: Value`	Single data point.
`Row`	`metric: str`, `labels: list[Label]`, `data_point: DataPoint`	Complete write record.
`MetricSeries`	`name: str`, `labels: list[Label]`	Series identity.
`SeriesPoints`	`series: MetricSeries`, `points: list[DataPoint]`	Series with its data.
`LabeledDataPoints`	`labels: list[Label]`, `data_points: list[DataPoint]`	Labels with points (from `select_all`).
`QueryOptions`	`labels`, `start`, `end`, `aggregation`, `downsample`, `limit`, `offset`	Advanced query configuration.
`DownsampleOptions`	`interval: int`	Downsample bucket width.
`SeriesSelection`	`metric: Optional[str]`, `matchers`, `start`, `end`	Series filter.
`SeriesMatcher`	`name: str`, `op: SeriesMatcherOp`, `value: str`	Single label matcher.
`RollupPolicy`	`id`, `metric`, `match_labels`, `interval`, `aggregation`, `bucket_origin`	Rollup definition.
`WriteResult`	`acknowledgement: WriteAcknowledgement`	Write durability level.
`DeleteSeriesResult`	`matched_series: int`, `tombstones_applied: int`	Deletion outcome.

Enums

Python type	Variants
`Value`	`F64`, `I64`, `U64`, `BOOL`, `BYTES`, `STR`, `HISTOGRAM`
`Aggregation`	`NONE`, `SUM`, `MIN`, `MAX`, `AVG`, `FIRST`, `LAST`, `COUNT`, `MEDIAN`, `RANGE`, `VARIANCE`, `STD_DEV`
`TimestampPrecision`	`NANOSECONDS`, `MICROSECONDS`, `MILLISECONDS`, `SECONDS`
`StorageRuntimeMode`	`READ_WRITE`, `COMPUTE_ONLY`
`RemoteSegmentCachePolicy`	`METADATA_ONLY`
`WalSyncMode`	`PER_APPEND`, `PERIODIC(interval)`
`WalReplayMode`	`STRICT`, `SALVAGE`
`WriteAcknowledgement`	`VOLATILE`, `APPENDED`, `DURABLE`
`SeriesMatcherOp`	`EQUAL`, `NOT_EQUAL`, `REGEX_MATCH`, `REGEX_NO_MATCH`
`HistogramCount`	`INT(v)`, `FLOAT(v)`
`HistogramResetHint`	`UNKNOWN`, `YES`, `NO`, `GAUGE`

FilesExpand file tree

python-bindings.md

Latest commit

History