Add scheduled refresh for tables that change at a fixed time of day

This commit is contained in:
2026-07-30 11:28:20 +02:00
parent 4a86b2282f
commit 0026a4df22
12 changed files with 652 additions and 66 deletions
+20
View File
@@ -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 ## [1.16.0] - 2026-06-11
### Added ### Added
+51 -8
View File
@@ -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. - **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. - **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). - 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 ```python
engine.refresh() # also reloads any expired TTL tables on demand 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 ## 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: 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. - **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. - 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=`) ## 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: 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 ```python
from sqlmem import CachingEngine, TableSpec, Delta, TTL from sqlmem import CachingEngine, TableSpec, Delta, TTL, Schedule
engine = CachingEngine( engine = CachingEngine(
base_engine, base_engine,
@@ -263,6 +299,11 @@ engine = CachingEngine(
refresh=TTL(seconds=1800), refresh=TTL(seconds=1800),
preload=True, preload=True,
), ),
TableSpec(
name="VW_NIGHTLY_PRICES",
refresh=Schedule("03:00"),
preload=True,
),
], ],
pragmas={"mmap_size": 32 * 1024**3, "page_size": 8192}, 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`. - **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. - **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. - `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=`/`indexes=`/`datetime_columns=` kwargs work exactly as before (lazy mode, no fail-fast). Passing both raises `ValueError`. - **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 ## 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.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.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.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 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 | | `loading` | a full load is in progress |
| `refreshing` | an incremental (delta) refresh is in progress | | `refreshing` | an incremental (delta) refresh is in progress |
| `ready` | cached and idle (up to date) | | `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 | | `error` | the last load failed |
| `tracking` | Meaning | | `tracking` | Meaning |
|---|---| |---|---|
| `delta` | kept in sync incrementally via a change column | | `delta` | kept in sync incrementally via a change column |
| `ttl` | full-reloaded when older than its TTL | | `ttl` | full-reloaded when older than its TTL |
| `schedule` | full-reloaded at fixed times of day |
| `static` | loaded on demand, never auto-refreshed | | `static` | loaded on demand, never auto-refreshed |
## Memory and very large tables ## 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_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_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_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 | | `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: 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] **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] **`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] **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 ## Dependencies
+2 -1
View File
@@ -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] **`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] **`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] **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 ## TODO — budoucí funkce
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "sqlmem" name = "sqlmem"
version = "1.16.0" version = "1.17.0"
description = "" description = ""
authors = [ authors = [
{name = "jan.doubravsky@gmail.com"} {name = "jan.doubravsky@gmail.com"}
+2 -1
View File
@@ -8,7 +8,7 @@ from .config import DEBUG
from .delta import DeltaConfig from .delta import DeltaConfig
from .engine import CachingEngine from .engine import CachingEngine
from .exceptions import ReadOnlyError, UndeclaredError, UnsupportedQueryError from .exceptions import ReadOnlyError, UndeclaredError, UnsupportedQueryError
from .spec import TTL, Delta, TableSpec from .spec import TTL, Delta, Schedule, TableSpec
from .stats import Stats, TableStats from .stats import Stats, TableStats
_DEFAULT_FORMAT = ( _DEFAULT_FORMAT = (
@@ -62,6 +62,7 @@ __all__ = [
"DeltaConfig", "DeltaConfig",
"Delta", "Delta",
"TTL", "TTL",
"Schedule",
"TableSpec", "TableSpec",
"ReadOnlyError", "ReadOnlyError",
"UnsupportedQueryError", "UnsupportedQueryError",
+9 -3
View File
@@ -325,15 +325,21 @@ class CacheManager:
).fetchone() ).fetchone()
return bool(row and row[0]) return bool(row and row[0])
def seconds_since_refresh(self, table: str) -> float | None: def last_refresh_at(self, table: str) -> datetime | None:
"""Age of a cached table in seconds, or None if it is not cached.""" """When a cached table was last (re)loaded, or None if it is not cached."""
with self._lock: with self._lock:
row = self._conn.execute( row = self._conn.execute(
"SELECT last_refresh_at FROM _sqlmem_tables WHERE table_name = ?", (table,) "SELECT last_refresh_at FROM _sqlmem_tables WHERE table_name = ?", (table,)
).fetchone() ).fetchone()
if not row or not row[0]: if not row or not row[0]:
return None 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() return (datetime.now(timezone.utc) - last).total_seconds()
def discover_columns(self, table: str, source_conn: sqlite3.Connection) -> list[str]: def discover_columns(self, table: str, source_conn: sqlite3.Connection) -> list[str]:
+92 -27
View File
@@ -1,5 +1,6 @@
import threading import threading
from dataclasses import replace from dataclasses import replace
from itertools import combinations
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -22,7 +23,8 @@ from .exceptions import UndeclaredError
from .executor import QueryExecutor from .executor import QueryExecutor
from .parser import Params, ParsedQuery, parse from .parser import Params, ParsedQuery, parse
from .registry import ColumnRegistry 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 from .stats import Stats, StatsCollector, TableState, TableStats
@@ -31,18 +33,21 @@ def _specs_to_config(
) -> tuple[ ) -> tuple[
dict[str, DeltaConfig], dict[str, DeltaConfig],
dict[str, int], dict[str, int],
dict[str, ScheduleSpec],
dict[str, list[str | list[str]]], dict[str, list[str | list[str]]],
dict[str, list[str]], dict[str, list[str]],
dict[str, list[str] | None], dict[str, list[str] | None],
]: ]:
"""Convert declarative ``TableSpec``s into the engine's internal config dicts. """Convert declarative ``TableSpec``s into the engine's internal config dicts.
Returns ``(delta, ttl, indexes, datetime_columns, declared)`` — the first four Returns ``(delta, ttl, schedule, indexes, datetime_columns, declared)`` — the
mirror the legacy kwargs; ``declared`` maps each table to its allowed columns first five mirror the legacy kwargs; ``declared`` maps each table to its
(``None`` = whole table / any column) for fail-fast query checking. allowed columns (``None`` = whole table / any column) for fail-fast query
checking.
""" """
delta: dict[str, DeltaConfig] = {} delta: dict[str, DeltaConfig] = {}
ttl: dict[str, int] = {} ttl: dict[str, int] = {}
schedule: dict[str, ScheduleSpec] = {}
indexes: dict[str, list[str | list[str]]] = {} indexes: dict[str, list[str | list[str]]] = {}
datetime_columns: dict[str, list[str]] = {} datetime_columns: dict[str, list[str]] = {}
declared: dict[str, list[str] | None] = {} declared: dict[str, list[str] | None] = {}
@@ -57,9 +62,11 @@ def _specs_to_config(
refresh = spec.refresh refresh = spec.refresh
if isinstance(refresh, TTL): if isinstance(refresh, TTL):
ttl[spec.name] = refresh.seconds ttl[spec.name] = refresh.seconds
elif isinstance(refresh, Schedule):
schedule[spec.name] = refresh.times
elif isinstance(refresh, DeltaConfig): elif isinstance(refresh, DeltaConfig):
delta[spec.name] = refresh delta[spec.name] = refresh
return delta, ttl, indexes, datetime_columns, declared return delta, ttl, schedule, indexes, datetime_columns, declared
class _LazySource: class _LazySource:
@@ -96,6 +103,7 @@ class CachingEngine:
source_engine: Engine, source_engine: Engine,
delta: dict[str, DeltaConfig] | None = None, delta: dict[str, DeltaConfig] | None = None,
ttl: dict[str, int] | None = None, ttl: dict[str, int] | None = None,
schedule: dict[str, ScheduleSpec] | None = None,
indexes: dict[str, list[str | list[str]]] | None = None, indexes: dict[str, list[str | list[str]]] | None = None,
in_memory: bool | None = None, in_memory: bool | None = None,
cache_db_path: str | Path | None = None, cache_db_path: str | Path | None = None,
@@ -112,17 +120,19 @@ class CachingEngine:
self._source_engine = source_engine self._source_engine = source_engine
# Declarative mode: a list of TableSpecs is converted to the same internal # Declarative mode: a list of TableSpecs is converted to the same internal
# config the legacy delta=/ttl=/indexes=/datetime_columns= kwargs produce, # config the legacy delta=/ttl=/schedule=/indexes=/datetime_columns= kwargs
# plus a declared-columns allowlist (for fail-fast) and preload set. # produce, plus a declared-columns allowlist (for fail-fast) and preload set.
self._declared: dict[str, list[str] | None] | None = None self._declared: dict[str, list[str] | None] | None = None
self._preload_specs: list[TableSpec] = [] self._preload_specs: list[TableSpec] = []
if tables is not None: 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( raise ValueError(
"Pass either tables=[TableSpec(...)] or the legacy " "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] self._preload_specs = [s for s in tables if s.preload]
use_memory = IN_MEMORY if in_memory is None else in_memory use_memory = IN_MEMORY if in_memory is None else in_memory
@@ -144,18 +154,13 @@ class CachingEngine:
self._stats = StatsCollector() self._stats = StatsCollector()
self._delta = self._resolve_delta(delta or {}) self._delta = self._resolve_delta(delta or {})
self._ttl = dict(ttl 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._index_columns = self._register_indexes(indexes or {})
self._refresher = DeltaRefresher(self._cache, self._delta) self._refresher = DeltaRefresher(self._cache, self._delta)
overlap = set(self._delta) & set(self._ttl) self._check_one_method_per_table()
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."
)
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 # Startup work (preload of declared tables + delta/TTL catch-up for
# tables restored from disk) can take a while on a cold start. By # tables restored from disk) can take a while on a cold start. By
# default it runs on the background thread so it never blocks # default it runs on the background thread so it never blocks
@@ -168,6 +173,24 @@ class CachingEngine:
logger.info("CachingEngine initialized.") 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( def _register_indexes(
self, indexes: dict[str, list[str | list[str]]] self, indexes: dict[str, list[str | list[str]]]
) -> dict[str, list[str]]: ) -> dict[str, list[str]]:
@@ -227,19 +250,19 @@ class CachingEngine:
errors: dict[str, TableError], errors: dict[str, TableError],
last_runs: dict[str, str], last_runs: dict[str, str],
) -> TableStats: ) -> 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: if name in self._delta:
tracking = "delta" tracking = "delta"
elif name in self._ttl: elif name in self._ttl:
tracking = "ttl" tracking = "ttl"
elif name in self._schedule:
tracking = "schedule"
else: else:
tracking = "static" tracking = "static"
state = table_stats.state state = table_stats.state
if state == TableState.READY and name in self._ttl: if state == TableState.READY and self._is_stale(name):
age = self._cache.seconds_since_refresh(name) state = TableState.STALE
if age is not None and age > self._ttl[name]:
state = TableState.STALE
last_refresh = last_runs.get(name) last_refresh = last_runs.get(name)
err = errors.get(name) err = errors.get(name)
@@ -255,6 +278,15 @@ class CachingEngine:
) )
return replace(table_stats, tracking=tracking, state=state, last_refresh=last_refresh) 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: def _make_executor(self, source: Any) -> QueryExecutor:
return QueryExecutor( return QueryExecutor(
self._cache, self._cache,
@@ -264,6 +296,7 @@ class CachingEngine:
self._delta, self._delta,
self._ttl, self._ttl,
self._index_columns, self._index_columns,
self._schedule,
) )
def _check_declared(self, parsed: ParsedQuery) -> None: def _check_declared(self, parsed: ParsedQuery) -> None:
@@ -329,6 +362,7 @@ class CachingEngine:
raw_conn = sa_conn.connection.dbapi_connection raw_conn = sa_conn.connection.dbapi_connection
self._refresher.refresh(raw_conn) self._refresher.refresh(raw_conn)
self._refresh_ttl(raw_conn) self._refresh_ttl(raw_conn)
self._refresh_scheduled(raw_conn)
except Exception as e: except Exception as e:
logger.error(f"Refresh cycle failed: {e}") logger.error(f"Refresh cycle failed: {e}")
@@ -341,26 +375,57 @@ class CachingEngine:
if age is None or age <= ttl: if age is None or age <= ttl:
continue continue
try: try:
columns = self._cache.get_table_columns(table) self._reload(table, source_conn)
full = self._cache.is_table_full(table)
self._cache.load_table(table, columns, source_conn, full=full)
logger.info(f"TTL refresh {table!r}: reloaded (age {age:.0f}s > {ttl}s)") logger.info(f"TTL refresh {table!r}: reloaded (age {age:.0f}s > {ttl}s)")
except Exception as e: except Exception as e:
logger.error(f"TTL refresh failed for {table!r}: {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 _start_refresh_thread(self, initial_catch_up: bool = True) -> None:
def loop() -> None: def loop() -> None:
if initial_catch_up: if initial_catch_up:
self._preload() # off-main-thread declared-table preload self._preload() # off-main-thread declared-table preload
self._run_refresh() # off-main-thread startup catch-up self._run_refresh() # off-main-thread startup catch-up
event = threading.Event() event = threading.Event()
while not event.wait(self._refresh_interval): while not event.wait(self._next_tick()):
self._run_refresh() self._run_refresh()
t = threading.Thread(target=loop, daemon=True, name="sqlmem-delta") t = threading.Thread(target=loop, daemon=True, name="sqlmem-delta")
t.start() t.start()
logger.debug(f"Delta refresh thread started (interval={self._refresh_interval}s)") 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: def invalidate(self, table: str) -> None:
logger.info(f"Manually invalidating cache for table {table!r}") logger.info(f"Manually invalidating cache for table {table!r}")
with self._cache._lock: with self._cache._lock:
+34 -20
View File
@@ -1,4 +1,5 @@
from collections.abc import Callable from collections.abc import Callable
from datetime import time
from typing import Any from typing import Any
from loguru import logger from loguru import logger
@@ -7,6 +8,7 @@ from .cache import CacheManager
from .delta import ResolvedDelta from .delta import ResolvedDelta
from .parser import ParsedQuery from .parser import ParsedQuery
from .registry import ColumnRegistry from .registry import ColumnRegistry
from .schedule import is_due, now_local
from .stats import StatsCollector from .stats import StatsCollector
@@ -20,6 +22,7 @@ class QueryExecutor:
delta: dict[str, ResolvedDelta] | None = None, delta: dict[str, ResolvedDelta] | None = None,
ttl: dict[str, int] | None = None, ttl: dict[str, int] | None = None,
index_columns: dict[str, list[str]] | None = None, index_columns: dict[str, list[str]] | None = None,
schedule: dict[str, tuple[time, ...]] | None = None,
) -> None: ) -> None:
self._cache = cache self._cache = cache
self._registry = registry self._registry = registry
@@ -28,14 +31,25 @@ class QueryExecutor:
self._delta = delta or {} self._delta = delta or {}
self._ttl = ttl or {} self._ttl = ttl or {}
self._index_columns = index_columns or {} self._index_columns = index_columns or {}
self._schedule = schedule or {}
def _ttl_expired(self, table: str) -> bool: def _stale_reason(self, table: str) -> str | None:
"""True if *table* has a TTL and its cached copy is older than that TTL.""" """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) ttl = self._ttl.get(table)
if ttl is None: if ttl is not None:
return False age = self._cache.seconds_since_refresh(table)
age = self._cache.seconds_since_refresh(table) if age is not None and age > ttl:
return 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]: def execute(self, parsed: ParsedQuery) -> list[dict]:
for table in parsed.tables: for table in parsed.tables:
@@ -63,31 +77,31 @@ class QueryExecutor:
self._ensure_columns(table, parsed.columns_by_table[table]) self._ensure_columns(table, parsed.columns_by_table[table])
def _full_satisfied(self, table: str) -> bool: 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 ( return (
self._cache.is_table_cached(table) self._cache.is_table_cached(table)
and self._cache.is_table_full(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: def _columns_satisfied(self, table: str, columns: list[str]) -> bool:
"""True if *table* is cached with all *columns* present and not TTL-expired.""" """True if *table* is cached with all *columns* present and not expired."""
if not self._cache.is_table_cached(table) or self._ttl_expired(table): if not self._cache.is_table_cached(table) or self._stale_reason(table) is not None:
return False return False
return set(columns).issubset(self._cache.get_table_columns(table)) return set(columns).issubset(self._cache.get_table_columns(table))
def _ensure_full(self, table: str) -> None: def _ensure_full(self, table: str) -> None:
"""Load every column of *table* (SELECT * / t.*), refetching unless already full.""" """Load every column of *table* (SELECT * / t.*), refetching unless already full."""
cached = self._cache.is_table_cached(table) 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}") logger.debug(f"Cache hit (full): {table!r}")
self._stats.record_hit() self._stats.record_hit()
return return
if cached and stale: if reason:
logger.info(f"Cache expired (ttl) — reloading {table!r} in full.") logger.info(f"Cache expired ({reason}) — reloading {table!r} in full.")
self._stats.record_refetch() self._stats.record_refetch()
elif cached: elif cached:
logger.warning(f"Re-fetching {table!r} in full — SELECT * requested.") 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)) self._load(table, columns, full=True, satisfied=lambda cols: self._full_satisfied(table))
def _ensure_columns(self, table: str, columns: list[str]) -> None: 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) missing = self._registry.needs_refetch(table, columns)
table_cached = self._cache.is_table_cached(table) 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}") logger.debug(f"Cache hit: {table!r} columns={columns}")
self._stats.record_hit() self._stats.record_hit()
return return
if stale: if reason:
logger.info(f"Cache expired (ttl) — reloading {table!r}.") logger.info(f"Cache expired ({reason}) — reloading {table!r}.")
self._stats.record_refetch() self._stats.record_refetch()
elif table_cached and missing: elif table_cached and missing:
logger.warning( logger.warning(
@@ -122,7 +136,7 @@ class QueryExecutor:
self._stats.record_miss() self._stats.record_miss()
all_columns = list(self._registry.get_columns(table)) + missing 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) full = table_cached and self._cache.is_table_full(table)
self._load( self._load(
table, table,
+87
View File
@@ -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)
+21 -3
View File
@@ -10,6 +10,7 @@ kwargs keep working; ``tables=`` is converted to the same internal config.
from dataclasses import dataclass, field from dataclasses import dataclass, field
from .delta import DeltaConfig from .delta import DeltaConfig
from .schedule import ScheduleSpec, parse_schedule
# Friendly alias for the declarative API; ``Delta`` and ``DeltaConfig`` are the # Friendly alias for the declarative API; ``Delta`` and ``DeltaConfig`` are the
# same type (``change_column`` + ``key_columns``), so either may be used as a # same type (``change_column`` + ``key_columns``), so either may be used as a
@@ -24,6 +25,22 @@ class TTL:
seconds: int 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("<Schedule>", self.times) # eager validation; table name unknown yet
@dataclass(frozen=True) @dataclass(frozen=True)
class TableSpec: class TableSpec:
"""Declarative specification of one cached table. """Declarative specification of one cached table.
@@ -33,8 +50,9 @@ class TableSpec:
a query asking for a column outside the list raises a query asking for a column outside the list raises
:class:`~sqlmem.exceptions.UndeclaredError`. :class:`~sqlmem.exceptions.UndeclaredError`.
*refresh* is a :class:`Delta` (change-column incremental sync) or :class:`TTL` *refresh* is a :class:`Delta` (change-column incremental sync), :class:`TTL`
(time-based full reload), or ``None`` for a static table loaded once. (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 *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 first query is a cache hit instead of paying a cold load; a copy already fresh
@@ -44,6 +62,6 @@ class TableSpec:
name: str name: str
columns: list[str] | None = None columns: list[str] | None = None
indexes: list[str | list[str]] = field(default_factory=list) 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) datetime_columns: list[str] = field(default_factory=list)
preload: bool = False preload: bool = False
+2 -2
View File
@@ -9,7 +9,7 @@ class TableState:
LOADING = "loading" # a full load is in progress LOADING = "loading" # a full load is in progress
REFRESHING = "refreshing" # an incremental (delta) refresh is in progress REFRESHING = "refreshing" # an incremental (delta) refresh is in progress
READY = "ready" # cached and idle 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 ERROR = "error" # the last load failed
@@ -25,7 +25,7 @@ class TableStats:
# the first cycle runs after start. Answers "is the refresh loop alive?". # the first cycle runs after start. Answers "is the refresh loop alive?".
last_refresh: str | None = None last_refresh: str | None = None
state: str = TableState.READY 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`` # 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. # resets to 0 on the next success, so > 0 means the table is currently failing.
last_error: str | None = None last_error: str | None = None
+331
View File
@@ -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