Reference

Public API

class timedb.TimeDBClient(ch_url: str | None = None)

Bases: object

__init__(ch_url: str | None = None)

create() → None

Create the series_values table and run_series mapping.

delete() → None

Drop both CH tables.

read_run_series(*, series_id: int) → list[int]

Return run_ids that touched a given series_id, latest first.

Data only — the energydb.runs PG table hydrates the metadata.

timedb.RETENTION_TIERS = frozenset({'forever', 'long', 'medium', 'short'})

frozenset() -> empty frozenset object frozenset(iterable) -> frozenset object

Build an immutable unordered collection of unique elements.

Profiling helpers

A lightweight phase-timer used by the read/write paths. Useful when diagnosing slow queries or large bulk inserts.

Opt-in per-phase timing collector for TimeDB internal operations.

Disabled by default — zero overhead when disabled (no perf_counter calls, no function calls in the hot path). Benchmark scripts activate it per-trial to collect phase-level timing breakdowns.

Not thread-safe; designed for single-threaded benchmark use.

Usage:

from timedb import profiling

profiling.enable() profiling.reset() # … run operation … phases = profiling.collect() # dict of phase -> elapsed seconds profiling.disable()

# Or, for hot-path instrumentation: with profiling._phase(profiling.PHASE_EDB_RESOLVE):

…

timedb.profiling.collect() → dict[str, float]

Return a copy of accumulated timings (in seconds).

timedb.profiling.disable() → None

Disable profiling and clear timings.

timedb.profiling.enable() → None

Enable profiling collection. Call before each trial.

timedb.profiling.is_enabled() → bool

Return True if profiling is currently active.

timedb.profiling.reset() → None

Clear accumulated timings while keeping profiling enabled.