diff --git a/CHANGELOG.md b/CHANGELOG.md index 7fda6e5..84c83d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,26 @@ All notable changes to this project will be documented in this file. --- +## [1.17.0] - 2026-07-30 + +### Added +- **Scheduled refresh at a fixed time of day** — `CachingEngine(engine, schedule={"VW_X": "03:00", "VW_Y": ["06:30", "14:30"]})` fully reloads a table at the times of day you name (`"HH:MM"` / `"HH:MM:SS"` strings or `datetime.time`, one or a list), for tables that change on a clock (nightly batch, morning import) rather than continuously or via a change column. Times are interpreted in the local timezone; an invalid or timezone-aware time raises `ValueError` at construction. + - **Slot-based, not timer-based** — a table is due when it hasn't been reloaded since its most recent scheduled moment (judged against the persisted `last_refresh_at`), so a slot missed while the process was down is caught up on the first refresh check after start, and a slot never fires twice. + - The background thread **shortens its tick** to land within a second of each slot instead of waiting up to `SQLMEM_REFRESH_INTERVAL`. + - **Read-time guarantee**, same as TTL: a query touching a table whose slot has passed reloads it before answering. + - Also available as a **`TableSpec.refresh=Schedule(...)`** strategy in declarative mode, alongside `Delta`/`TTL`; `Schedule` validates its times eagerly at construction. + - `stats` reports `tracking="schedule"`, and `state="stale"` once a slot has passed. +- New public type `Schedule`; `schedule.py` module with the parsing/normalization (`parse_schedule`) and slot maths (`previous_occurrence`, `next_occurrence`, `seconds_until_next`, `is_due`) it and the engine share. +- `CacheManager.last_refresh_at(table)` — the load timestamp as a `datetime` (`seconds_since_refresh` now derives from it). + +### Changed +- `pyproject.toml` — bumped version to `1.17.0`. +- **Refresh methods are mutually exclusive** — a table listed under any two of `delta`, `ttl`, `schedule` raises `ValueError` (was: only the `delta`/`ttl` pair). Per `TableSpec` this was already structurally impossible (one `refresh` field). +- `QueryExecutor`'s TTL staleness check (`_ttl_expired`) is generalized to `_stale_reason`, covering TTL and schedule expiry identically wherever staleness is checked — read-time reload, and the double-checked-locking `satisfied` predicates. +- `CachingEngine._refresh_ttl`/new `_refresh_scheduled` share a `_reload` helper. + +--- + ## [1.16.0] - 2026-06-11 ### Added diff --git a/README.md b/README.md index bde50c5..03e91f0 100644 --- a/README.md +++ b/README.md @@ -212,12 +212,48 @@ engine = CachingEngine( - **Read-time guarantee** — when a query touches a TTL table whose cache is older than its TTL, the table is fully reloaded *before* the query is answered, so a stale copy is never returned. - **Proactive** — the background thread also full-reloads expired TTL tables every `SQLMEM_REFRESH_INTERVAL` seconds, keeping them warm so reads usually don't pay the reload latency. - TTL age is measured from `last_refresh_at`, which is persisted in `cache.db`, so the guarantee holds across restarts (an expired table is reloaded on first use after start). -- A table may be in **either** `delta` **or** `ttl`, not both (delta already keeps it fresh) — supplying both raises `ValueError`. +- A table has **exactly one** refresh method — `delta`, `ttl` or [`schedule`](#scheduled-refresh-at-a-fixed-time-of-day); listing it under two of them raises `ValueError`. ```python engine.refresh() # also reloads any expired TTL tables on demand ``` +## Scheduled refresh (at a fixed time of day) + +When a table only changes at a known moment — an overnight batch, a nightly import, a report generated every morning — a TTL wastes reloads waiting for the next tick to notice. Give it a **schedule** instead: the table is fully reloaded at the times of day you name. + +```python +from datetime import time +from sqlmem import CachingEngine + +engine = CachingEngine( + base_engine, + schedule={ + "VW_NIGHTLY_PRICES": "03:00", # every day at 03:00 + "VW_STOCK_LEVELS": ["06:30", "14:30"], # twice a day + "VW_REPORT": time(5, 15), # datetime.time works too + }, +) +``` + +Times are `"HH:MM"` / `"HH:MM:SS"` strings or `datetime.time` objects, one or a list of them, interpreted in the **local timezone** of the process. An unparsable or timezone-aware time raises `ValueError` at construction. + +- **Slot-based, not timer-based** — a table is due when it has not been reloaded since its most recent scheduled moment. Consequences: + - **A missed slot is caught up.** If the process was down at 03:00 and starts at 09:00, the table reloads on the first refresh check after start (`last_refresh_at` is persisted in `cache.db`, so this holds across restarts). + - **A slot never fires twice**, however often the check runs. +- **Fires on time** — the background thread shortens its wait so it wakes within about a second of each slot, instead of up to `SQLMEM_REFRESH_INTERVAL` late. +- **Read-time guarantee** — same as TTL: a query touching a table whose slot has passed triggers the reload *before* it is answered, so a pre-batch copy is never returned even if the background thread hasn't run yet. +- Around a daylight-saving change a slot may land an hour off for that one day; the reload still happens. + +Also available as a `TableSpec.refresh` strategy in [declarative mode](#declarative-initialization-tables): + +```python +from sqlmem import TableSpec, Schedule +TableSpec("VW_NIGHTLY_PRICES", refresh=Schedule("03:00"), preload=True) +``` + +Pick `schedule` when the source changes on a clock, `ttl` when it changes continuously but a bounded lag is acceptable, and `delta` when the table is large and has a change column. + ## Secondary indexes To accelerate lookups, you can declare **secondary indexes** per table — they are created on the in-memory SQLite copy so `WHERE`/`JOIN` filters on those columns run as indexed searches instead of full scans: @@ -236,14 +272,14 @@ Each value is a list of index definitions: a string is a single-column index, a - **Index columns are pulled into the cache automatically** (like delta key columns), so the index exists from the first load even if your queries don't select those columns. - Indexes are **recreated after every (re)load** — full loads, TTL reloads, and `invalidate()` + re-fetch all rebuild them — so they're always present, and they persist in `cache.db` across restarts. -- Delta-tracked tables already get a unique index on their key columns; secondary indexes are independent and can be combined with `delta` or `ttl`. +- Delta-tracked tables already get a unique index on their key columns; secondary indexes are independent and can be combined with `delta`, `ttl` or `schedule`. ## Declarative initialization (`tables=`) Instead of the lazy "learn columns from queries" mode, you can **declare every table up front** with `tables=[TableSpec(...)]` — its columns, indexes, refresh strategy and which columns are datetimes — and have the engine preload them and reject anything undeclared: ```python -from sqlmem import CachingEngine, TableSpec, Delta, TTL +from sqlmem import CachingEngine, TableSpec, Delta, TTL, Schedule engine = CachingEngine( base_engine, @@ -263,6 +299,11 @@ engine = CachingEngine( refresh=TTL(seconds=1800), preload=True, ), + TableSpec( + name="VW_NIGHTLY_PRICES", + refresh=Schedule("03:00"), + preload=True, + ), ], pragmas={"mmap_size": 32 * 1024**3, "page_size": 8192}, ) @@ -270,8 +311,8 @@ engine = CachingEngine( - **Preload** — `preload=True` tables are loaded at startup (on the background thread by default, so startup isn't blocked; pass `blocking_startup_refresh=True` to load them synchronously before serving). A copy already fresh in the persistent cache is **skipped**, so a warm restart is instant. During warm-up a table reports `TableState.LOADING` in [`stats`](#runtime-statistics) — handy for gating a `503` until it's `ready`. - **Fail-fast** — a query for a table without a `TableSpec`, or for a column outside a spec's declared `columns` (including `SELECT *` on a column-restricted table), raises `UndeclaredError` instead of silently kicking off an expensive lazy load. Use `columns=None` to cache the whole table and allow any column. -- `refresh=` takes a `Delta(change_column=…, key_columns=…)` (same as `DeltaConfig`) or `TTL(seconds=…)`, or `None` for a static table. -- **Backward compatible** — omit `tables=` and the legacy `delta=`/`ttl=`/`indexes=`/`datetime_columns=` kwargs work exactly as before (lazy mode, no fail-fast). Passing both raises `ValueError`. +- `refresh=` takes a `Delta(change_column=…, key_columns=…)` (same as `DeltaConfig`), `TTL(seconds=…)` or [`Schedule(times)`](#scheduled-refresh-at-a-fixed-time-of-day), or `None` for a static table. +- **Backward compatible** — omit `tables=` and the legacy `delta=`/`ttl=`/`schedule=`/`indexes=`/`datetime_columns=` kwargs work exactly as before (lazy mode, no fail-fast). Passing both raises `ValueError`. ## Persistence @@ -350,7 +391,7 @@ engine.invalidate("orders") # drop one table from cache; next query re-fetches engine.reset() # wipe the whole cache (RAM + cache.db) — full clean slate engine.hard_reset() # disk mode: delete the file and reopen with current pragmas/page_size engine.vacuum() # disk mode: incremental VACUUM (reclaim free pages from delta churn) -engine.refresh() # pull deltas for all delta-tracked tables now +engine.refresh() # pull deltas + reload any expired ttl/schedule tables now engine.close() # flush to disk and shut down background thread ``` @@ -389,13 +430,14 @@ Each `TableStats` reports a live processing **state** and how the table is kept | `loading` | a full load is in progress | | `refreshing` | an incremental (delta) refresh is in progress | | `ready` | cached and idle (up to date) | -| `stale` | a TTL table whose cache has expired; reloads on next access | +| `stale` | a TTL table past its max age, or a scheduled table whose slot has passed; reloads on next access | | `error` | the last load failed | | `tracking` | Meaning | |---|---| | `delta` | kept in sync incrementally via a change column | | `ttl` | full-reloaded when older than its TTL | +| `schedule` | full-reloaded at fixed times of day | | `static` | loaded on demand, never auto-refreshed | ## Memory and very large tables @@ -419,7 +461,7 @@ Set via environment variables or a `.env` file: | `SQLMEM_IN_MEMORY` | `true` | `false` queries `cache.db` on disk directly (no RAM copy); overridden by the `in_memory` constructor arg | | `SQLMEM_BACKUP_INTERVAL` | `3600` | Disk backup interval in seconds (in-memory mode only) | | `SQLMEM_SQL_DIALECT` | `tsql` | sqlglot dialect used to parse incoming SQL (e.g. `tsql`, `postgres`, `mysql`) | -| `SQLMEM_REFRESH_INTERVAL` | `300` | background refresh tick (seconds) — delta pulls and proactive TTL reloads | +| `SQLMEM_REFRESH_INTERVAL` | `300` | background refresh tick (seconds) — delta pulls and proactive TTL/schedule reloads; shortened automatically when a `schedule` slot falls sooner | | `SQLMEM_FETCH_BATCH` | `10000` | rows fetched per batch when loading a table — caps peak memory for huge tables | Most of these can also be passed **per engine** to the constructor, overriding the env default — handy for running two engines (with separate cache files) in one process, and for tests: @@ -483,6 +525,7 @@ Set `SQLMEM_DEBUG=true` in `.env` to make the default level DEBUG when no explic - [x] **Primary-key auto-discovery** from the source DB (`inspect(engine).get_pk_constraint`) so `key_columns` is only needed for views. - [x] **`engine.reset()`** — wipe RAM + `cache.db` for a clean rebuild after structural changes. - [x] **Per-table TTL** (time-to-live) — bounded-staleness full refresh for tables without a change column. +- [x] **Scheduled refresh** — full reload at fixed times of day for tables that change on a clock (nightly batch, morning import). ## Dependencies diff --git a/project.md b/project.md index af66e53..eb4c5b9 100644 --- a/project.md +++ b/project.md @@ -225,7 +225,8 @@ SQLMEM_DEBUG=true # DEBUG level — podrobný výpis každého dotazu, cache o - [x] **`vacuum(incremental=True)` varuje bez `auto_vacuum=INCREMENTAL`**: dřív tichý no-op; teď zaloguje warning (a jak to opravit) a vrátí se. - [x] **`Stats.db_size_bytes`**: velikost cache souboru na disku (0 v memory módu) ve `stats` pro monitoring. - [x] **Ochrana proti cache stampede**: `load_table` dělá double-checked locking — po získání `_load_lock` znovu ověří, zda tabulku mezitím nenahrál souběžný loader (cached + sloupce + ne-stale), a redundantní reload přeskočí. Bez toho druhý dotaz během studeného loadu velké tabulky spustil druhý plný reload (212M řádků = +2 h). -- [x] **Deklarativní inicializace (`tables=[TableSpec(...)]`)**: předem se deklaruje každá tabulka (`TableSpec(name, columns, indexes, refresh=Delta(...)|TTL(...), datetime_columns, preload)`). `preload=True` se na pozadí (nebo blokujícně) přednahraje při startu; co je v persistentní cache čerstvé, se přeskočí (warm restart = instant). Nedeklarovaná tabulka/sloupec (i `SELECT *` na sloupcově omezené tabulce) → `UndeclaredError` (fail-fast) místo tichého líného loadu; `columns=None` = celá tabulka + libovolný sloupec. Plně zpětně kompatibilní: bez `tables=` fungují staré `delta=/ttl=/indexes=/datetime_columns=` jako dřív (kombinace obojího → `ValueError`). +- [x] **Deklarativní inicializace (`tables=[TableSpec(...)]`)**: předem se deklaruje každá tabulka (`TableSpec(name, columns, indexes, refresh=Delta(...)|TTL(...)|Schedule(...), datetime_columns, preload)`). `preload=True` se na pozadí (nebo blokujícně) přednahraje při startu; co je v persistentní cache čerstvé, se přeskočí (warm restart = instant). Nedeklarovaná tabulka/sloupec (i `SELECT *` na sloupcově omezené tabulce) → `UndeclaredError` (fail-fast) místo tichého líného loadu; `columns=None` = celá tabulka + libovolný sloupec. Plně zpětně kompatibilní: bez `tables=` fungují staré `delta=/ttl=/schedule=/indexes=/datetime_columns=` jako dřív (kombinace obojího → `ValueError`). +- [x] **Plánovaný refresh v konkrétní čas (`schedule=`)**: `schedule={"VW_X": "03:00"}` (nebo seznam časů / `datetime.time`) — full reload v zadané časy dne v lokální timezone, pro tabulky měněné dávkou v pevnou hodinu (noční batch, ranní import). Validace (neplatný/tz-aware čas → `ValueError`) proběhne hned při konstrukci. Tabulka je „due", pokud se od posledního plánovaného slotu nenačetla (`is_due` porovnává proti `last_refresh_at`), takže zmeškaný slot se dohoní po startu a jeden slot nikdy nespustí reload dvakrát; read-time garance jako u TTL. Background thread si zkrátí čekání (`_next_tick`), aby slot spustil do ~sekundy. Dostupné i jako `TableSpec.refresh=Schedule(...)` v deklarativním módu. Tabulka má vždy jen jednu metodu obnovy — `delta` / `ttl` / `schedule`, jinak `ValueError` (`_check_one_method_per_table`). ## TODO — budoucí funkce diff --git a/pyproject.toml b/pyproject.toml index b22713f..b9c8288 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "sqlmem" -version = "1.16.0" +version = "1.17.0" description = "" authors = [ {name = "jan.doubravsky@gmail.com"} diff --git a/src/sqlmem/__init__.py b/src/sqlmem/__init__.py index 73538e7..4515a9c 100644 --- a/src/sqlmem/__init__.py +++ b/src/sqlmem/__init__.py @@ -8,7 +8,7 @@ from .config import DEBUG from .delta import DeltaConfig from .engine import CachingEngine from .exceptions import ReadOnlyError, UndeclaredError, UnsupportedQueryError -from .spec import TTL, Delta, TableSpec +from .spec import TTL, Delta, Schedule, TableSpec from .stats import Stats, TableStats _DEFAULT_FORMAT = ( @@ -62,6 +62,7 @@ __all__ = [ "DeltaConfig", "Delta", "TTL", + "Schedule", "TableSpec", "ReadOnlyError", "UnsupportedQueryError", diff --git a/src/sqlmem/cache.py b/src/sqlmem/cache.py index 934389b..0becd99 100644 --- a/src/sqlmem/cache.py +++ b/src/sqlmem/cache.py @@ -325,15 +325,21 @@ class CacheManager: ).fetchone() return bool(row and row[0]) - def seconds_since_refresh(self, table: str) -> float | None: - """Age of a cached table in seconds, or None if it is not cached.""" + def last_refresh_at(self, table: str) -> datetime | None: + """When a cached table was last (re)loaded, or None if it is not cached.""" with self._lock: row = self._conn.execute( "SELECT last_refresh_at FROM _sqlmem_tables WHERE table_name = ?", (table,) ).fetchone() if not row or not row[0]: return None - last = datetime.fromisoformat(row[0]) + return datetime.fromisoformat(row[0]) + + def seconds_since_refresh(self, table: str) -> float | None: + """Age of a cached table in seconds, or None if it is not cached.""" + last = self.last_refresh_at(table) + if last is None: + return None return (datetime.now(timezone.utc) - last).total_seconds() def discover_columns(self, table: str, source_conn: sqlite3.Connection) -> list[str]: diff --git a/src/sqlmem/engine.py b/src/sqlmem/engine.py index 192b904..42b61c7 100644 --- a/src/sqlmem/engine.py +++ b/src/sqlmem/engine.py @@ -1,5 +1,6 @@ import threading from dataclasses import replace +from itertools import combinations from pathlib import Path from typing import Any @@ -22,7 +23,8 @@ from .exceptions import UndeclaredError from .executor import QueryExecutor from .parser import Params, ParsedQuery, parse from .registry import ColumnRegistry -from .spec import TTL, TableSpec +from .schedule import ScheduleSpec, is_due, now_local, parse_schedule, seconds_until_next +from .spec import TTL, Schedule, TableSpec from .stats import Stats, StatsCollector, TableState, TableStats @@ -31,18 +33,21 @@ def _specs_to_config( ) -> tuple[ dict[str, DeltaConfig], dict[str, int], + dict[str, ScheduleSpec], dict[str, list[str | list[str]]], dict[str, list[str]], dict[str, list[str] | None], ]: """Convert declarative ``TableSpec``s into the engine's internal config dicts. - Returns ``(delta, ttl, indexes, datetime_columns, declared)`` — the first four - mirror the legacy kwargs; ``declared`` maps each table to its allowed columns - (``None`` = whole table / any column) for fail-fast query checking. + Returns ``(delta, ttl, schedule, indexes, datetime_columns, declared)`` — the + first five mirror the legacy kwargs; ``declared`` maps each table to its + allowed columns (``None`` = whole table / any column) for fail-fast query + checking. """ delta: dict[str, DeltaConfig] = {} ttl: dict[str, int] = {} + schedule: dict[str, ScheduleSpec] = {} indexes: dict[str, list[str | list[str]]] = {} datetime_columns: dict[str, list[str]] = {} declared: dict[str, list[str] | None] = {} @@ -57,9 +62,11 @@ def _specs_to_config( refresh = spec.refresh if isinstance(refresh, TTL): ttl[spec.name] = refresh.seconds + elif isinstance(refresh, Schedule): + schedule[spec.name] = refresh.times elif isinstance(refresh, DeltaConfig): delta[spec.name] = refresh - return delta, ttl, indexes, datetime_columns, declared + return delta, ttl, schedule, indexes, datetime_columns, declared class _LazySource: @@ -96,6 +103,7 @@ class CachingEngine: source_engine: Engine, delta: dict[str, DeltaConfig] | None = None, ttl: dict[str, int] | None = None, + schedule: dict[str, ScheduleSpec] | None = None, indexes: dict[str, list[str | list[str]]] | None = None, in_memory: bool | None = None, cache_db_path: str | Path | None = None, @@ -112,17 +120,19 @@ class CachingEngine: self._source_engine = source_engine # Declarative mode: a list of TableSpecs is converted to the same internal - # config the legacy delta=/ttl=/indexes=/datetime_columns= kwargs produce, - # plus a declared-columns allowlist (for fail-fast) and preload set. + # config the legacy delta=/ttl=/schedule=/indexes=/datetime_columns= kwargs + # produce, plus a declared-columns allowlist (for fail-fast) and preload set. self._declared: dict[str, list[str] | None] | None = None self._preload_specs: list[TableSpec] = [] if tables is not None: - if any(x is not None for x in (delta, ttl, indexes, datetime_columns)): + if any(x is not None for x in (delta, ttl, schedule, indexes, datetime_columns)): raise ValueError( "Pass either tables=[TableSpec(...)] or the legacy " - "delta=/ttl=/indexes=/datetime_columns= kwargs, not both." + "delta=/ttl=/schedule=/indexes=/datetime_columns= kwargs, not both." ) - delta, ttl, indexes, datetime_columns, self._declared = _specs_to_config(tables) + delta, ttl, schedule, indexes, datetime_columns, self._declared = _specs_to_config( + tables + ) self._preload_specs = [s for s in tables if s.preload] use_memory = IN_MEMORY if in_memory is None else in_memory @@ -144,18 +154,13 @@ class CachingEngine: self._stats = StatsCollector() self._delta = self._resolve_delta(delta or {}) self._ttl = dict(ttl or {}) + self._schedule = {t: parse_schedule(t, s) for t, s in (schedule or {}).items()} self._index_columns = self._register_indexes(indexes or {}) self._refresher = DeltaRefresher(self._cache, self._delta) - overlap = set(self._delta) & set(self._ttl) - if overlap: - raise ValueError( - f"Tables {sorted(overlap)} are in both delta and ttl — a table is " - "either delta-refreshed (has a change column) or TTL-refreshed (full " - "reload), not both." - ) + self._check_one_method_per_table() - if self._delta or self._ttl or self._preload_specs: + if self._delta or self._ttl or self._schedule or self._preload_specs: # Startup work (preload of declared tables + delta/TTL catch-up for # tables restored from disk) can take a while on a cold start. By # default it runs on the background thread so it never blocks @@ -168,6 +173,24 @@ class CachingEngine: logger.info("CachingEngine initialized.") + def _check_one_method_per_table(self) -> None: + """Reject a table configured with more than one refresh method. + + Per-table this is already impossible via ``TableSpec.refresh`` (one + field); this guards the legacy kwargs, where ``delta=``/``ttl=``/ + ``schedule=`` are independent dicts a caller could overlap by mistake. + """ + methods = {"delta": set(self._delta), "ttl": set(self._ttl), "schedule": set(self._schedule)} + for (name_a, tables_a), (name_b, tables_b) in combinations(methods.items(), 2): + overlap = tables_a & tables_b + if overlap: + raise ValueError( + f"Tables {sorted(overlap)} are in both {name_a} and {name_b} — a table " + "has exactly one refresh method: delta (incremental, via a change " + "column), ttl (full reload once older than a max age) or schedule " + "(full reload at fixed times of day)." + ) + def _register_indexes( self, indexes: dict[str, list[str | list[str]]] ) -> dict[str, list[str]]: @@ -227,19 +250,19 @@ class CachingEngine: errors: dict[str, TableError], last_runs: dict[str, str], ) -> TableStats: - """Annotate a TableStats with refresh tracking, TTL staleness, errors and run time.""" + """Annotate a TableStats with refresh tracking, staleness, errors and run time.""" if name in self._delta: tracking = "delta" elif name in self._ttl: tracking = "ttl" + elif name in self._schedule: + tracking = "schedule" else: tracking = "static" state = table_stats.state - if state == TableState.READY and name in self._ttl: - age = self._cache.seconds_since_refresh(name) - if age is not None and age > self._ttl[name]: - state = TableState.STALE + if state == TableState.READY and self._is_stale(name): + state = TableState.STALE last_refresh = last_runs.get(name) err = errors.get(name) @@ -255,6 +278,15 @@ class CachingEngine: ) return replace(table_stats, tracking=tracking, state=state, last_refresh=last_refresh) + def _is_stale(self, name: str) -> bool: + """True if a cached table is past its TTL or its last scheduled reload time.""" + if name in self._ttl: + age = self._cache.seconds_since_refresh(name) + return age is not None and age > self._ttl[name] + if name in self._schedule: + return is_due(self._schedule[name], self._cache.last_refresh_at(name), now_local()) + return False + def _make_executor(self, source: Any) -> QueryExecutor: return QueryExecutor( self._cache, @@ -264,6 +296,7 @@ class CachingEngine: self._delta, self._ttl, self._index_columns, + self._schedule, ) def _check_declared(self, parsed: ParsedQuery) -> None: @@ -329,6 +362,7 @@ class CachingEngine: raw_conn = sa_conn.connection.dbapi_connection self._refresher.refresh(raw_conn) self._refresh_ttl(raw_conn) + self._refresh_scheduled(raw_conn) except Exception as e: logger.error(f"Refresh cycle failed: {e}") @@ -341,26 +375,57 @@ class CachingEngine: if age is None or age <= ttl: continue try: - columns = self._cache.get_table_columns(table) - full = self._cache.is_table_full(table) - self._cache.load_table(table, columns, source_conn, full=full) + self._reload(table, source_conn) logger.info(f"TTL refresh {table!r}: reloaded (age {age:.0f}s > {ttl}s)") except Exception as e: logger.error(f"TTL refresh failed for {table!r}: {e}") + def _refresh_scheduled(self, source_conn: Any) -> None: + """Full-reload scheduled tables whose scheduled time of day has passed.""" + now = now_local() + for table, times in self._schedule.items(): + if not self._cache.is_table_cached(table): + continue + if not is_due(times, self._cache.last_refresh_at(table), now): + continue + try: + self._reload(table, source_conn) + logger.info(f"Scheduled refresh {table!r}: reloaded (slot passed)") + except Exception as e: + logger.error(f"Scheduled refresh failed for {table!r}: {e}") + + def _reload(self, table: str, source_conn: Any) -> None: + """Full-reload a cached table, keeping its cached column set and full status.""" + columns = self._cache.get_table_columns(table) + full = self._cache.is_table_full(table) + self._cache.load_table(table, columns, source_conn, full=full) + def _start_refresh_thread(self, initial_catch_up: bool = True) -> None: def loop() -> None: if initial_catch_up: self._preload() # off-main-thread declared-table preload self._run_refresh() # off-main-thread startup catch-up event = threading.Event() - while not event.wait(self._refresh_interval): + while not event.wait(self._next_tick()): self._run_refresh() t = threading.Thread(target=loop, daemon=True, name="sqlmem-delta") t.start() logger.debug(f"Delta refresh thread started (interval={self._refresh_interval}s)") + def _next_tick(self) -> float: + """Seconds to sleep before the next refresh cycle. + + Normally the fixed refresh interval, but shortened when a scheduled + table's time of day falls sooner, so a schedule fires within a second + of its slot instead of waiting for the next tick. + """ + if not self._schedule: + return float(self._refresh_interval) + # +1s so the wake-up lands just past the slot, never a hair before it. + soonest = min(seconds_until_next(t, now_local()) for t in self._schedule.values()) + 1.0 + return max(1.0, min(float(self._refresh_interval), soonest)) + def invalidate(self, table: str) -> None: logger.info(f"Manually invalidating cache for table {table!r}") with self._cache._lock: diff --git a/src/sqlmem/executor.py b/src/sqlmem/executor.py index 4f6cd22..b034bf8 100644 --- a/src/sqlmem/executor.py +++ b/src/sqlmem/executor.py @@ -1,4 +1,5 @@ from collections.abc import Callable +from datetime import time from typing import Any from loguru import logger @@ -7,6 +8,7 @@ from .cache import CacheManager from .delta import ResolvedDelta from .parser import ParsedQuery from .registry import ColumnRegistry +from .schedule import is_due, now_local from .stats import StatsCollector @@ -20,6 +22,7 @@ class QueryExecutor: delta: dict[str, ResolvedDelta] | None = None, ttl: dict[str, int] | None = None, index_columns: dict[str, list[str]] | None = None, + schedule: dict[str, tuple[time, ...]] | None = None, ) -> None: self._cache = cache self._registry = registry @@ -28,14 +31,25 @@ class QueryExecutor: self._delta = delta or {} self._ttl = ttl or {} self._index_columns = index_columns or {} + self._schedule = schedule or {} - def _ttl_expired(self, table: str) -> bool: - """True if *table* has a TTL and its cached copy is older than that TTL.""" + def _stale_reason(self, table: str) -> str | None: + """Why *table*'s cached copy must be reloaded before answering a query. + + Returns ``"ttl"``/``"schedule"`` or ``None`` (fresh) — the single check + both ``_ensure_full``/``_ensure_columns`` and the double-checked-locking + ``satisfied`` predicates use, so TTL and schedule expiry are treated + identically everywhere staleness matters. + """ ttl = self._ttl.get(table) - if ttl is None: - return False - age = self._cache.seconds_since_refresh(table) - return age is not None and age > ttl + if ttl is not None: + age = self._cache.seconds_since_refresh(table) + if age is not None and age > ttl: + return "ttl" + times = self._schedule.get(table) + if times is not None and is_due(times, self._cache.last_refresh_at(table), now_local()): + return "schedule" + return None def execute(self, parsed: ParsedQuery) -> list[dict]: for table in parsed.tables: @@ -63,31 +77,31 @@ class QueryExecutor: self._ensure_columns(table, parsed.columns_by_table[table]) def _full_satisfied(self, table: str) -> bool: - """True if *table* is cached in full and not TTL-expired (a SELECT * hit).""" + """True if *table* is cached in full and not expired (a SELECT * hit).""" return ( self._cache.is_table_cached(table) and self._cache.is_table_full(table) - and not self._ttl_expired(table) + and self._stale_reason(table) is None ) def _columns_satisfied(self, table: str, columns: list[str]) -> bool: - """True if *table* is cached with all *columns* present and not TTL-expired.""" - if not self._cache.is_table_cached(table) or self._ttl_expired(table): + """True if *table* is cached with all *columns* present and not expired.""" + if not self._cache.is_table_cached(table) or self._stale_reason(table) is not None: return False return set(columns).issubset(self._cache.get_table_columns(table)) def _ensure_full(self, table: str) -> None: """Load every column of *table* (SELECT * / t.*), refetching unless already full.""" cached = self._cache.is_table_cached(table) - stale = cached and self._ttl_expired(table) + reason = self._stale_reason(table) if cached else None - if cached and self._cache.is_table_full(table) and not stale: + if cached and self._cache.is_table_full(table) and not reason: logger.debug(f"Cache hit (full): {table!r}") self._stats.record_hit() return - if cached and stale: - logger.info(f"Cache expired (ttl) — reloading {table!r} in full.") + if reason: + logger.info(f"Cache expired ({reason}) — reloading {table!r} in full.") self._stats.record_refetch() elif cached: logger.warning(f"Re-fetching {table!r} in full — SELECT * requested.") @@ -99,18 +113,18 @@ class QueryExecutor: self._load(table, columns, full=True, satisfied=lambda cols: self._full_satisfied(table)) def _ensure_columns(self, table: str, columns: list[str]) -> None: - """Load *table* with at least *columns*, refetching on new columns or TTL expiry.""" + """Load *table* with at least *columns*, refetching on new columns or expiry.""" missing = self._registry.needs_refetch(table, columns) table_cached = self._cache.is_table_cached(table) - stale = table_cached and self._ttl_expired(table) + reason = self._stale_reason(table) if table_cached else None - if table_cached and not missing and not stale: + if table_cached and not missing and not reason: logger.debug(f"Cache hit: {table!r} columns={columns}") self._stats.record_hit() return - if stale: - logger.info(f"Cache expired (ttl) — reloading {table!r}.") + if reason: + logger.info(f"Cache expired ({reason}) — reloading {table!r}.") self._stats.record_refetch() elif table_cached and missing: logger.warning( @@ -122,7 +136,7 @@ class QueryExecutor: self._stats.record_miss() all_columns = list(self._registry.get_columns(table)) + missing - # Preserve a fully-cached table's status across a TTL reload. + # Preserve a fully-cached table's status across a TTL/schedule reload. full = table_cached and self._cache.is_table_full(table) self._load( table, diff --git a/src/sqlmem/schedule.py b/src/sqlmem/schedule.py new file mode 100644 index 0000000..c5ca717 --- /dev/null +++ b/src/sqlmem/schedule.py @@ -0,0 +1,87 @@ +from datetime import datetime, time, timedelta, timezone + +ScheduleSpec = str | time | list[str | time] | tuple[str | time, ...] +"""One or more times of day, as ``"HH:MM"`` / ``"HH:MM:SS"`` strings or :class:`~datetime.time`.""" + + +def parse_schedule(table: str, spec: ScheduleSpec) -> tuple[time, ...]: + """Normalize a per-table schedule into a sorted tuple of times of day. + + Accepts a single time or a list of them; raises ``ValueError`` on an empty + spec or an unparsable / timezone-aware time. + """ + if isinstance(spec, (str, time)): + items: list[str | time] = [spec] + elif isinstance(spec, (list, tuple)): + items = list(spec) + else: + raise ValueError( + f"Invalid schedule {spec!r} for {table!r} — expected a time of day " + "('HH:MM', 'HH:MM:SS' or datetime.time), or a list of them." + ) + if not items: + raise ValueError( + f"Schedule for {table!r} is empty — give at least one time of day, e.g. '03:00'." + ) + return tuple(sorted({_parse_time(table, item) for item in items})) + + +def _parse_time(table: str, value: str | time) -> time: + if isinstance(value, time): + parsed = value + else: + try: + parsed = time.fromisoformat(value) + except (TypeError, ValueError): + raise ValueError( + f"Invalid schedule time {value!r} for {table!r} — " + "expected 'HH:MM' or 'HH:MM:SS'." + ) from None + if parsed.tzinfo is not None: + raise ValueError( + f"Schedule time {value!r} for {table!r} carries a timezone — " + "schedule times are interpreted in the local timezone, give them bare." + ) + return parsed.replace(microsecond=0) + + +def now_local() -> datetime: + """Current time as a timezone-aware datetime in the local timezone.""" + return datetime.now(timezone.utc).astimezone() + + +def previous_occurrence(times: tuple[time, ...], now: datetime) -> datetime: + """The most recent scheduled moment at or before *now*.""" + return max(c for c in _candidates(times, now, days=(0, -1)) if c <= now) + + +def next_occurrence(times: tuple[time, ...], now: datetime) -> datetime: + """The first scheduled moment strictly after *now*.""" + return min(c for c in _candidates(times, now, days=(0, 1)) if c > now) + + +def _candidates( + times: tuple[time, ...], now: datetime, days: tuple[int, ...] +) -> list[datetime]: + return [ + datetime.combine(now.date() + timedelta(days=d), t, tzinfo=now.tzinfo) + for d in days + for t in times + ] + + +def seconds_until_next(times: tuple[time, ...], now: datetime) -> float: + return (next_occurrence(times, now) - now).total_seconds() + + +def is_due(times: tuple[time, ...], last_refresh: datetime | None, now: datetime) -> bool: + """True if the table has not been reloaded since its last scheduled moment. + + Comparing against the *last scheduled occurrence* (rather than tracking fired + timers) makes the schedule survive restarts and coarse refresh ticks: a table + whose 03:00 slot was missed because the process was down reloads on the first + check after start, and never fires twice for the same slot. + """ + if last_refresh is None: # not cached — the first query loads it anyway + return False + return last_refresh < previous_occurrence(times, now) diff --git a/src/sqlmem/spec.py b/src/sqlmem/spec.py index d2d96e2..8b8df75 100644 --- a/src/sqlmem/spec.py +++ b/src/sqlmem/spec.py @@ -10,6 +10,7 @@ kwargs keep working; ``tables=`` is converted to the same internal config. from dataclasses import dataclass, field from .delta import DeltaConfig +from .schedule import ScheduleSpec, parse_schedule # Friendly alias for the declarative API; ``Delta`` and ``DeltaConfig`` are the # same type (``change_column`` + ``key_columns``), so either may be used as a @@ -24,6 +25,22 @@ class TTL: seconds: int +@dataclass(frozen=True) +class Schedule: + """Time-of-day refresh strategy: full-reload the table at fixed times of day. + + *times* is one time of day or a list of them (``"HH:MM"``/``"HH:MM:SS"`` + strings or :class:`datetime.time`), interpreted in the local timezone. + Validated eagerly so a typo surfaces at construction, not at the first + refresh tick. + """ + + times: ScheduleSpec + + def __post_init__(self) -> None: + parse_schedule("", self.times) # eager validation; table name unknown yet + + @dataclass(frozen=True) class TableSpec: """Declarative specification of one cached table. @@ -33,8 +50,9 @@ class TableSpec: a query asking for a column outside the list raises :class:`~sqlmem.exceptions.UndeclaredError`. - *refresh* is a :class:`Delta` (change-column incremental sync) or :class:`TTL` - (time-based full reload), or ``None`` for a static table loaded once. + *refresh* is a :class:`Delta` (change-column incremental sync), :class:`TTL` + (time-based full reload) or :class:`Schedule` (full reload at fixed times of + day), or ``None`` for a static table loaded once. *preload=True* loads the table at startup (in the background by default) so the first query is a cache hit instead of paying a cold load; a copy already fresh @@ -44,6 +62,6 @@ class TableSpec: name: str columns: list[str] | None = None indexes: list[str | list[str]] = field(default_factory=list) - refresh: DeltaConfig | TTL | None = None + refresh: DeltaConfig | TTL | Schedule | None = None datetime_columns: list[str] = field(default_factory=list) preload: bool = False diff --git a/src/sqlmem/stats.py b/src/sqlmem/stats.py index b45a413..bbc6b9a 100644 --- a/src/sqlmem/stats.py +++ b/src/sqlmem/stats.py @@ -9,7 +9,7 @@ class TableState: LOADING = "loading" # a full load is in progress REFRESHING = "refreshing" # an incremental (delta) refresh is in progress READY = "ready" # cached and idle - STALE = "stale" # TTL expired — will reload on next access + STALE = "stale" # TTL expired / schedule slot passed — reloads on next access ERROR = "error" # the last load failed @@ -25,7 +25,7 @@ class TableStats: # the first cycle runs after start. Answers "is the refresh loop alive?". last_refresh: str | None = None state: str = TableState.READY - tracking: str = "static" # "delta" | "ttl" | "static" + tracking: str = "static" # "delta" | "ttl" | "schedule" | "static" # Most recent load/refresh failure for this table, if any. ``consecutive_failures`` # resets to 0 on the next success, so > 0 means the table is currently failing. last_error: str | None = None diff --git a/tests/test_schedule.py b/tests/test_schedule.py new file mode 100644 index 0000000..fb9daf4 --- /dev/null +++ b/tests/test_schedule.py @@ -0,0 +1,331 @@ +import sqlite3 +from datetime import datetime, time, timedelta, timezone + +import pytest +from sqlalchemy import create_engine + +import sqlmem.engine as eng_mod +from sqlmem import CachingEngine, Delta, DeltaConfig, Schedule, TableSpec +from sqlmem.cache import CacheManager +from sqlmem.executor import QueryExecutor +from sqlmem.parser import parse +from sqlmem.registry import ColumnRegistry +from sqlmem.schedule import ( + is_due, + next_occurrence, + parse_schedule, + previous_occurrence, + seconds_until_next, +) +from sqlmem.stats import StatsCollector, TableState + +# --------------------------------------------------------------------------- +# schedule.py — spec parsing +# --------------------------------------------------------------------------- + + +def test_parse_single_time_string(): + assert parse_schedule("t", "03:00") == (time(3, 0),) + + +def test_parse_accepts_seconds_and_time_objects(): + assert parse_schedule("t", "03:30:15") == (time(3, 30, 15),) + assert parse_schedule("t", time(6, 45)) == (time(6, 45),) + + +def test_parse_list_is_sorted_and_deduplicated(): + assert parse_schedule("t", ["15:30", "03:00", "03:00"]) == (time(3, 0), time(15, 30)) + + +@pytest.mark.parametrize("spec", ["3pm", "25:00", "", "03:00:00.5x", None, 300]) +def test_parse_rejects_invalid_time(spec): + with pytest.raises(ValueError): + parse_schedule("t", spec) + + +def test_parse_rejects_empty_list(): + with pytest.raises(ValueError): + parse_schedule("t", []) + + +def test_parse_rejects_timezone_aware_time(): + with pytest.raises(ValueError): + parse_schedule("t", time(3, 0, tzinfo=timezone.utc)) + + +# --------------------------------------------------------------------------- +# schedule.py — occurrence maths +# --------------------------------------------------------------------------- + +TZ = timezone(timedelta(hours=2)) +TIMES = (time(3, 0), time(15, 30)) + + +def at(hour, minute=0, day=10): + return datetime(2026, 6, day, hour, minute, tzinfo=TZ) + + +def test_previous_occurrence_earlier_today(): + assert previous_occurrence(TIMES, at(16)) == at(15, 30) + + +def test_previous_occurrence_falls_back_to_yesterday(): + assert previous_occurrence(TIMES, at(2)) == at(15, 30, day=9) + + +def test_previous_occurrence_includes_the_exact_moment(): + assert previous_occurrence(TIMES, at(3)) == at(3) + + +def test_next_occurrence_and_seconds_until(): + assert next_occurrence(TIMES, at(4)) == at(15, 30) + assert next_occurrence(TIMES, at(16)) == at(3, 0, day=11) + assert seconds_until_next(TIMES, at(15, 0)) == 30 * 60 + + +def test_is_due_only_when_last_refresh_predates_the_slot(): + now = at(16) + assert is_due(TIMES, at(15, 0), now) is True # loaded before the 15:30 slot + assert is_due(TIMES, at(15, 45), now) is False # loaded after it + assert is_due(TIMES, None, now) is False # not cached at all + + +def test_is_due_compares_across_timezones(): + # last_refresh is persisted in UTC; the slot is local. + assert is_due(TIMES, datetime(2026, 6, 10, 12, 0, tzinfo=timezone.utc), at(16)) is True + assert is_due(TIMES, datetime(2026, 6, 10, 14, 0, tzinfo=timezone.utc), at(16)) is False + + +# --------------------------------------------------------------------------- +# executor-level: read-time guarantee +# --------------------------------------------------------------------------- + + +@pytest.fixture +def source_conn(): + conn = sqlite3.connect(":memory:") + conn.executescript( + """ + CREATE TABLE products (id TEXT, name TEXT, price TEXT); + INSERT INTO products VALUES ('1', 'Widget', '9.99'), ('2', 'Gadget', '19.99'); + """ + ) + conn.commit() + yield conn + conn.close() + + +def make_executor(tmp_path, source_conn, schedule): + cache = CacheManager(db_path=tmp_path / "cache.db", backup_interval=9999) + registry = ColumnRegistry(cache.connection) + stats = StatsCollector() + return QueryExecutor(cache, registry, source_conn, stats, schedule=schedule) + + +def run(executor, sql, params=None): + return executor.execute(parse(sql, params)) + + +def slot(minutes_from_now): + """A schedule time of day *minutes_from_now* relative to the real clock.""" + when = datetime.now().astimezone() + timedelta(minutes=minutes_from_now) + return parse_schedule("t", when.time().replace(microsecond=0)) + + +def backdate(cache, table, hours=2): + """Pretend the cached copy was loaded *hours* ago, i.e. before a past slot. + + Only ``last_refresh_at`` is moved (the timestamp the schedule is judged + against), so the reload path is exercised against the real clock. + """ + stamp = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() + cache.connection.execute( + "UPDATE _sqlmem_tables SET last_refresh_at = ? WHERE table_name = ?", (stamp, table) + ) + cache.connection.commit() + + +def test_query_reloads_when_slot_has_passed(tmp_path, source_conn): + executor = make_executor(tmp_path, source_conn, {"products": slot(-1)}) + run(executor, "SELECT id, price FROM products") # miss → load + source_conn.execute("UPDATE products SET price = '1.11' WHERE id = '1'") + source_conn.commit() + backdate(executor._cache, "products") + + rows = {r["id"]: r for r in run(executor, "SELECT id, price FROM products")} + assert rows["1"]["price"] == "1.11" # slot passed → reloaded before answering + assert executor._stats.refetches == 1 + assert executor._stats.misses == 1 + + +def test_query_before_slot_is_cache_hit(tmp_path, source_conn): + # The slot is a minute away, so it cannot have passed since the load. + executor = make_executor(tmp_path, source_conn, {"products": slot(1)}) + run(executor, "SELECT id, price FROM products") + source_conn.execute("UPDATE products SET price = '1.11' WHERE id = '1'") + source_conn.commit() + + rows = {r["id"]: r for r in run(executor, "SELECT id, price FROM products")} + assert rows["1"]["price"] == "9.99" # still fresh → cached value served + assert executor._stats.hits == 1 + assert executor._stats.refetches == 0 + + +def test_scheduled_reload_preserves_full_status(tmp_path, source_conn): + executor = make_executor(tmp_path, source_conn, {"products": slot(-1)}) + run(executor, "SELECT * FROM products") # full load + backdate(executor._cache, "products") + run(executor, "SELECT * FROM products") # slot passed → full reload + assert executor._cache.is_table_full("products") is True + + +def test_reload_happens_once_per_slot(tmp_path, source_conn): + executor = make_executor(tmp_path, source_conn, {"products": slot(-1)}) + run(executor, "SELECT id, price FROM products") + backdate(executor._cache, "products") + run(executor, "SELECT id, price FROM products") # reload for that slot + run(executor, "SELECT id, price FROM products") # same slot → cache hit + assert executor._stats.refetches == 1 + assert executor._stats.hits == 1 + + +def test_untracked_table_never_expires_by_schedule(tmp_path, source_conn): + executor = make_executor(tmp_path, source_conn, {"other": slot(-1)}) + run(executor, "SELECT id, name FROM products") + backdate(executor._cache, "products") + rows = {r["id"]: r for r in run(executor, "SELECT id, name FROM products")} + assert rows["1"]["name"] == "Widget" + assert executor._stats.hits == 1 + + +# --------------------------------------------------------------------------- +# engine level: legacy schedule= kwarg +# --------------------------------------------------------------------------- + + +@pytest.fixture +def source_db(tmp_path): + db_path = tmp_path / "source.db" + conn = sqlite3.connect(db_path) + conn.executescript( + """ + CREATE TABLE products (id TEXT PRIMARY KEY, name TEXT, changed TEXT); + INSERT INTO products VALUES ('1', 'Widget', '2026-06-01 10:00:00'); + """ + ) + conn.commit() + conn.close() + return db_path + + +@pytest.fixture +def source_engine(source_db): + engine = create_engine(f"sqlite:///{source_db}") + yield engine + engine.dispose() + + +@pytest.fixture +def patched_cache(tmp_path, monkeypatch): + monkeypatch.setattr(eng_mod, "CACHE_DB_PATH", tmp_path / "cache.db") + monkeypatch.setattr(eng_mod, "BACKUP_INTERVAL_SECONDS", 9999) + + +def test_background_scheduled_refresh(source_engine, source_db, patched_cache): + engine = CachingEngine(source_engine, schedule={"products": slot(-1)[0]}) + engine.execute("SELECT id, name FROM products") + + conn = sqlite3.connect(source_db) + conn.execute("UPDATE products SET name = 'Widget2' WHERE id = '1'") + conn.commit() + conn.close() + + backdate(engine._cache, "products") + engine.refresh() # background-style reload of the table whose slot passed + rows = engine.execute("SELECT id, name FROM products") + assert rows[0]["name"] == "Widget2" + engine.close() + + +def test_stats_report_schedule_tracking_and_staleness(source_engine, patched_cache): + engine = CachingEngine(source_engine, schedule={"products": ["03:00", "15:30"]}) + engine.execute("SELECT id, name FROM products") + + stats = engine.stats + assert stats.tables["products"].tracking == "schedule" + assert stats.tables["products"].state == TableState.READY + + backdate(engine._cache, "products", hours=48) # both slots have since passed + assert engine.stats.tables["products"].state == TableState.STALE + engine.close() + + +def test_next_tick_shortens_to_the_next_slot(source_engine, patched_cache): + engine = CachingEngine(source_engine) # no schedule → plain interval + engine._refresh_interval = 300 + assert engine._next_tick() == 300 + + engine._schedule = {"products": slot(0.5)} + assert 1.0 < engine._next_tick() <= 32.0 + engine.close() + + +def test_invalid_schedule_rejected_at_construction(source_engine, patched_cache): + with pytest.raises(ValueError): + CachingEngine(source_engine, schedule={"products": "half past three"}) + + +@pytest.mark.parametrize( + "kwargs", + [ + {"ttl": {"products": 300}, "schedule": {"products": "03:00"}}, + { + "delta": {"products": DeltaConfig(change_column="changed", key_columns=["id"])}, + "schedule": {"products": "03:00"}, + }, + ], +) +def test_schedule_overlapping_another_method_raises(source_engine, patched_cache, kwargs): + with pytest.raises(ValueError): + CachingEngine(source_engine, **kwargs) + + +def test_tables_and_schedule_kwarg_are_mutually_exclusive(source_engine, patched_cache): + with pytest.raises(ValueError): + CachingEngine( + source_engine, + tables=[TableSpec("products", ["id"])], + schedule={"products": "03:00"}, + ) + + +# --------------------------------------------------------------------------- +# declarative mode: Schedule refresh strategy on TableSpec +# --------------------------------------------------------------------------- + + +def test_tablespec_schedule_tracking(source_engine, patched_cache): + engine = CachingEngine( + source_engine, + tables=[ + TableSpec( + "products", + ["id", "name"], + refresh=Schedule(["03:00", "15:30"]), + preload=True, + ) + ], + blocking_startup_refresh=True, + ) + assert engine.stats.tables["products"].tracking == "schedule" + engine.close() + + +def test_schedule_object_validates_eagerly(): + with pytest.raises(ValueError): + Schedule("not a time") + + +def test_schedule_and_delta_are_distinct_strategies(): + # Sanity check that Schedule isn't accidentally aliased to Delta/TTL. + assert Schedule is not Delta