diff --git a/docs/9_memoization.md b/docs/9_memoization.md index fd51c2b..2ff952f 100644 --- a/docs/9_memoization.md +++ b/docs/9_memoization.md @@ -83,19 +83,24 @@ db.memoize(func, data, ttl=datetime(2026, 1, 7)) ## Cache Maintenance -The `typedal.caching` module provides utilities for cache management: +The `typedal.caching` module provides utilities for cache management. +Every database with caching enabled has its own cache tables, so each function takes the database to act on: ```python -from typedal.caching import clear_cache, remove_cache_for_table, clear_expired +from typedal.caching import cache_models, clear_cache, remove_cache_for_table, clear_expired # Remove all cache entries -clear_cache() +clear_cache(db) # Invalidate all cache entries related to a specific table -remove_cache_for_table(User) +remove_cache_for_table(db, User) # Clean up expired entries only -clear_expired() +clear_expired(db) + +# The cache models bound to this database, e.g. to count entries: +Cache, CacheDependency = cache_models(db) +Cache.count() ``` You can also use the CLI: diff --git a/src/typedal/caching.py b/src/typedal/caching.py index 418ec35..536320c 100644 --- a/src/typedal/caching.py +++ b/src/typedal/caching.py @@ -11,7 +11,6 @@ import dill from .fields import TypedField -from .helpers import throw from .rows import TypedRows from .tables import TypedTable from .types import CacheStatus, Field, Query, Rows, Set @@ -60,6 +59,43 @@ class _TypedalCacheDependency(TypedTable): idx: TypedField[int] +CACHE_TABLE = "typedal_cache" +CACHE_DEPENDENCY_TABLE = "typedal_cache_dependency" + +CacheModels = tuple[type[_TypedalCache], type[_TypedalCacheDependency]] + + +def define_cache_models(db: "TypeDAL") -> CacheModels: + """ + Define the cache tables on `db`, using subclasses that belong to this database only. + + `db.define` binds the class it is given, so defining the module-level `_TypedalCache` itself on + every database would leave it bound to whichever database was created last (and `close()` on + any of them would unbind it for all). A fresh subclass per database keeps the bindings apart, + while the table names stay `typedal_cache` and `typedal_cache_dependency`. + """ + cache: type[_TypedalCache] = type("_TypedalCache", (_TypedalCache,), {"__module__": __name__}) + dependency: type[_TypedalCacheDependency] = type( + "_TypedalCacheDependency", (_TypedalCacheDependency,), {"__module__": __name__} + ) + return db.try_define(cache), db.try_define(dependency) + + +def cache_models(db: "TypeDAL") -> CacheModels: + """ + Get the cache models bound to `db`. + + Raises: + RuntimeError: when `db` was created with caching disabled. + """ + cache = db._class_map.get(CACHE_TABLE) + dependency = db._class_map.get(CACHE_DEPENDENCY_TABLE) + if cache is None or dependency is None: + raise RuntimeError("TypeDAL caching is not enabled for this database (enable_typedal_caching=False)!") + + return t.cast(type[_TypedalCache], cache), t.cast(type[_TypedalCacheDependency], dependency) + + def prepare(field: t.Any) -> str: """ Prepare data to be used in a cache key. @@ -157,64 +193,65 @@ def _determine_dependencies(instance: TypedRows[t.Any], rows: Rows, depends_on: return _get_dependency_ids(rows, dependency_keys) -def remove_cache(idx: int | t.Iterable[int], table: str) -> None: +def remove_cache(db: "TypeDAL", idx: int | t.Iterable[int], table: str) -> None: """ Remove any cache entries that are dependant on one or multiple indices of a table. """ if not isinstance(idx, t.Iterable): idx = [idx] - related = ( - _TypedalCacheDependency.where(table=table).where(lambda row: row.idx.belongs(idx)).select("entry").to_sql() - ) + cache, dependency = cache_models(db) + related = dependency.where(table=table).where(lambda row: row.idx.belongs(idx)).select("entry").to_sql() - _TypedalCache.where(_TypedalCache.id.belongs(related)).delete() + cache.where(cache.id.belongs(related)).delete() -def remove_cache_for_table(table: str) -> None: +def remove_cache_for_table(db: "TypeDAL", table: str | type[TypedTable]) -> None: """ Remove all cache entries that depend on a table. Used for inserts where we don't know which cached queries the new row would match. """ - related = _TypedalCacheDependency.where(table=table).select("entry").to_sql() - _TypedalCache.where(_TypedalCache.id.belongs(related)).delete() + if not isinstance(table, str): + table = str(table._table) + + cache, dependency = cache_models(db) + related = dependency.where(table=table).select("entry").to_sql() + cache.where(cache.id.belongs(related)).delete() -def clear_cache() -> None: +def clear_cache(db: "TypeDAL") -> None: """ Remove everything from the cache. Immediately commits """ - db: TypeDAL = _TypedalCache._db or throw( - RuntimeError("@define or db.define is not called on typedal caching classes yet!") - ) - - _TypedalCache.truncate("RESTART IDENTITY CASCADE") + cache, _ = cache_models(db) + cache.truncate("RESTART IDENTITY CASCADE") db.commit() -def clear_expired() -> int: +def clear_expired(db: "TypeDAL") -> int: """ Remove all expired items from the cache. By default, expired items are only removed when trying to access them. """ + cache, _ = cache_models(db) now = get_now() - return len(_TypedalCache.where(_TypedalCache.expires_at != None).where(_TypedalCache.expires_at < now).delete()) + return len(cache.where(cache.expires_at != None).where(cache.expires_at < now).delete()) -def _remove_cache(s: Set, tablename: str) -> None: +def _remove_cache(db: "TypeDAL", s: Set, tablename: str) -> None: """ Used as the table._before_delete for every TypeDAL table (on by default). """ indeces = s.select("id").column("id") - remove_cache(indeces, tablename) + remove_cache(db, indeces, tablename) -def _remove_cache_after_update(rows: Set, tablename: str) -> None: +def _remove_cache_after_update(db: "TypeDAL", rows: Set, tablename: str) -> None: """ Used as the table._after_update for every TypeDAL table (on by default). @@ -224,10 +261,10 @@ def _remove_cache_after_update(rows: Set, tablename: str) -> None: """ affected_ids = getattr(rows, "affected_ids", None) if affected_ids is None: - remove_cache_for_table(tablename) + remove_cache_for_table(db, tablename) else: # cache dependencies only exist for TypedTables, whose primary key is always the integer id - remove_cache(t.cast(list[int], affected_ids), tablename) + remove_cache(db, t.cast(list[int], affected_ids), tablename) def get_expire( @@ -253,6 +290,7 @@ def get_expire( def _insert_cache_entry( + db: "TypeDAL", key: str, data: t.Any, expires_at: dt.datetime | None, @@ -267,13 +305,14 @@ def _insert_cache_entry( into a half-applied one. The entry lands with the caller's own next commit instead - and is rolled back with everything else when there isn't one, which is the correct outcome for a cache. """ - entry = _TypedalCache.insert( + cache, dependency = cache_models(db) + entry = cache.insert( key=key, data=dill.dumps(data), expires_at=expires_at, ) - _TypedalCacheDependency.bulk_insert([{"entry": entry, "table": table, "idx": idx} for table, idx in deps]) + dependency.bulk_insert([{"entry": entry, "table": table, "idx": idx} for table, idx in deps]) def save_to_cache[T_TypedTable: TypedTable]( @@ -293,7 +332,7 @@ def save_to_cache[T_TypedTable: TypedTable]( if rows: expires_at = get_expire(expires_at=expires_at, ttl=ttl) or c.get("expires_at") deps = _determine_dependencies(instance, rows, c["depends_on"]) - _insert_cache_entry(key, instance, expires_at, deps) + _insert_cache_entry(instance.db, key, instance, expires_at, deps) instance.metadata["cache"]["status"] = "fresh" return instance @@ -317,7 +356,7 @@ def __bool__(self) -> bool: # pragma: no cover _CACHE_SENTINEL: t.Final[CacheMiss] = CacheMiss() -def _fetch_cached_payload(key: str) -> tuple[t.Any, t.Any] | None: +def _fetch_cached_payload(key: str, db: "TypeDAL") -> tuple[t.Any, t.Any] | None: """ Retrieves and validates a cache entry from the database. @@ -326,11 +365,13 @@ def _fetch_cached_payload(key: str) -> tuple[t.Any, t.Any] | None: Args: key: The unique string identifier for the cache entry. + db: The database whose cache tables to read. Returns: A tuple of (deserialized_data, db_row) if valid; None if missing or expired. """ - row = _TypedalCache.where(key=key).first() + cache, _ = cache_models(db) + row = cache.where(key=key).first() if not row: return None @@ -361,7 +402,7 @@ def _load_from_cache(key: str, db: "TypeDAL") -> t.Any | None: Returns: The re-hydrated model instance, or None if the load fails. """ - result = _fetch_cached_payload(key) + result = _fetch_cached_payload(key, db) if not result: return None @@ -378,7 +419,7 @@ def _load_from_cache(key: str, db: "TypeDAL") -> t.Any | None: return inst -def _load_memoize_from_cache(key: str) -> t.Any: +def _load_memoize_from_cache(key: str, db: "TypeDAL") -> t.Any: """ Low-level retrieval for memoized results. @@ -386,12 +427,13 @@ def _load_memoize_from_cache(key: str) -> t.Any: Args: key: Cache key. + db: The database whose cache tables to read. Returns: The deserialized data or the _CACHE_SENTINEL object if not found. """ with contextlib.suppress(Exception): - if result := _fetch_cached_payload(key): + if result := _fetch_cached_payload(key, db): return result[0] return _CACHE_SENTINEL @@ -434,14 +476,13 @@ def humanize_bytes(size: int | float) -> str: return f"{size:.2f} {suffixes[suffix_index]}" -def _expired_and_valid_query() -> tuple[str, str]: +def _expired_and_valid_query(db: "TypeDAL") -> tuple[str, str]: + cache, _ = cache_models(db) expired_items = ( - _TypedalCache.where(lambda row: (row.expires_at < get_now()) & (row.expires_at != None)) - .select(_TypedalCache.id) - .to_sql() + cache.where(lambda row: (row.expires_at < get_now()) & (row.expires_at != None)).select(cache.id).to_sql() ) - valid_items = _TypedalCache.where(~_TypedalCache.id.belongs(expired_items)).select(_TypedalCache.id).to_sql() + valid_items = cache.where(~cache.id.belongs(expired_items)).select(cache.id).to_sql() return expired_items, valid_items @@ -463,11 +504,12 @@ class Stats[T](t.TypedDict): def _row_stats(db: "TypeDAL", table: str, query: Query) -> RowStats: - count_field = _TypedalCacheDependency.entry.count() - stats: TypedRows[_TypedalCacheDependency] = db(query & (_TypedalCacheDependency.table == table)).select( - _TypedalCacheDependency.entry, + _, dependency = cache_models(db) + count_field = dependency.entry.count() + stats: TypedRows[_TypedalCacheDependency] = db(query & (dependency.table == table)).select( + dependency.entry, count_field, - groupby=_TypedalCacheDependency.entry, + groupby=dependency.entry, ) return { "Dependent Cache Entries": len(stats), @@ -478,14 +520,15 @@ def row_stats(db: "TypeDAL", table: str, row_id: str) -> Stats[RowStats]: """ Collect caching stats for a specific table row (by ID). """ - expired_items, valid_items = _expired_and_valid_query() + _, dependency = cache_models(db) + expired_items, valid_items = _expired_and_valid_query(db) - query = _TypedalCacheDependency.idx == row_id + query = dependency.idx == row_id return { "total": _row_stats(db, table, query), - "valid": _row_stats(db, table, _TypedalCacheDependency.entry.belongs(valid_items) & query), - "expired": _row_stats(db, table, _TypedalCacheDependency.entry.belongs(expired_items) & query), + "valid": _row_stats(db, table, dependency.entry.belongs(valid_items) & query), + "expired": _row_stats(db, table, dependency.entry.belongs(expired_items) & query), } @@ -499,11 +542,12 @@ def row_stats(db: "TypeDAL", table: str, row_id: str) -> Stats[RowStats]: def _table_stats(db: "TypeDAL", table: str, query: Query) -> TableStats: - count_field = _TypedalCacheDependency.entry.count() - stats: TypedRows[_TypedalCacheDependency] = db(query & (_TypedalCacheDependency.table == table)).select( - _TypedalCacheDependency.entry, + _, dependency = cache_models(db) + count_field = dependency.entry.count() + stats: TypedRows[_TypedalCacheDependency] = db(query & (dependency.table == table)).select( + dependency.entry, count_field, - groupby=_TypedalCacheDependency.entry, + groupby=dependency.entry, ) return { "Dependent Cache Entries": len(stats), @@ -515,12 +559,13 @@ def table_stats(db: "TypeDAL", table: str) -> Stats[TableStats]: """ Collect caching stats for a table. """ - expired_items, valid_items = _expired_and_valid_query() + _, dependency = cache_models(db) + expired_items, valid_items = _expired_and_valid_query(db) return { - "total": _table_stats(db, table, _TypedalCacheDependency.id > 0), - "valid": _table_stats(db, table, _TypedalCacheDependency.entry.belongs(valid_items)), - "expired": _table_stats(db, table, _TypedalCacheDependency.entry.belongs(expired_items)), + "total": _table_stats(db, table, dependency.id > 0), + "valid": _table_stats(db, table, dependency.entry.belongs(valid_items)), + "expired": _table_stats(db, table, dependency.entry.belongs(expired_items)), } @@ -535,14 +580,15 @@ def table_stats(db: "TypeDAL", table: str) -> Stats[TableStats]: def _calculate_stats(db: "TypeDAL", query: Query) -> GenericStats: - sum_len_field = _TypedalCache.data.len().sum() + cache, dependency = cache_models(db) + sum_len_field = cache.data.len().sum() size_row = db(query).select(sum_len_field).first() size = size_row[sum_len_field] if size_row else 0 return { - "entries": _TypedalCache.where(query).count(), - "dependencies": db(_TypedalCacheDependency.entry.belongs(query)).count(), + "entries": cache.where(query).count(), + "dependencies": db(dependency.entry.belongs(query)).count(), "size": humanize_bytes(size), } @@ -551,12 +597,13 @@ def calculate_stats(db: "TypeDAL") -> Stats[GenericStats]: """ Collect generic caching stats. """ - expired_items, valid_items = _expired_and_valid_query() + cache, _ = cache_models(db) + expired_items, valid_items = _expired_and_valid_query(db) return { - "total": _calculate_stats(db, _TypedalCache.id > 0), - "valid": _calculate_stats(db, _TypedalCache.id.belongs(valid_items)), - "expired": _calculate_stats(db, _TypedalCache.id.belongs(expired_items)), + "total": _calculate_stats(db, cache.id > 0), + "valid": _calculate_stats(db, cache.id.belongs(valid_items)), + "expired": _calculate_stats(db, cache.id.belongs(expired_items)), } @@ -605,7 +652,7 @@ def memoize[T: t.Any]( _, hashed_key = create_and_hash_cache_key(key, *[getattr(arg, "id", None) for arg in args], kwargs) # Try to load from cache - cached = _load_memoize_from_cache(hashed_key) + cached = _load_memoize_from_cache(hashed_key, db) if cached is not _CACHE_SENTINEL: return cached, "cached" # Cache miss - compute result @@ -643,6 +690,6 @@ def track_collect(qb: "QueryBuilder[t.Any]", _: TypedRows[t.Any], raw: Rows) -> else: expires_at = get_expire(ttl=ttl) - _insert_cache_entry(hashed_key, result, expires_at, deps) + _insert_cache_entry(db, hashed_key, result, expires_at, deps) return result, "fresh" diff --git a/src/typedal/cli.py b/src/typedal/cli.py index 0ca9bbf..da9c603 100644 --- a/src/typedal/cli.py +++ b/src/typedal/cli.py @@ -664,10 +664,10 @@ def cache_clear( db = TypeDAL(config=config, migrate=False, fake_migrate=False) if purge: - caching.clear_cache() + caching.clear_cache(db) print("Emptied cache") else: - n = caching.clear_expired() + n = caching.clear_expired(db) print(f"Removed {n} expired from cache") db.commit() diff --git a/src/typedal/core.py b/src/typedal/core.py index 92a93e5..eb88b1a 100644 --- a/src/typedal/core.py +++ b/src/typedal/core.py @@ -343,8 +343,7 @@ def __init__( install_update(self._adapter) if config.caching: - self.try_define(_TypedalCache) - self.try_define(_TypedalCacheDependency) + define_cache_models(self) def session(self) -> AsyncSession: """ @@ -407,7 +406,9 @@ def close(self) -> None: super().close() finally: for model in set(self._builder.class_map.values()): - model.unbind() + # a model defined on several databases is bound to the last one; leave that binding be: + if model._db is self: + model.unbind() self._builder.class_map.clear() if adapter is not None: # pragma: no branch - defensive cleanup @@ -757,7 +758,6 @@ def as_typescript(self, *tables: str | type[TypedTable]) -> str: from .updates import UpdateSet # noqa: E402 from .caching import ( # isort: skip # noqa: E402 + define_cache_models, memoize, - _TypedalCache, - _TypedalCacheDependency, ) diff --git a/src/typedal/define.py b/src/typedal/define.py index 3b63ccd..02d387c 100644 --- a/src/typedal/define.py +++ b/src/typedal/define.py @@ -131,9 +131,10 @@ def define[T: t.Any](self, cls: t.Type[T], **kwargs: t.Unpack[DefineKwargs]) -> if not tablename.startswith("typedal_") and cache_dependency: from .caching import _remove_cache, _remove_cache_after_update, remove_cache_for_table - table._after_insert.append(lambda _row, _id: remove_cache_for_table(tablename)) - table._after_update.append(lambda rows, _row: _remove_cache_after_update(rows, tablename)) - table._before_delete.append(lambda s: _remove_cache(s, tablename)) + db = self.db + table._after_insert.append(lambda _row, _id: remove_cache_for_table(db, tablename)) + table._after_update.append(lambda rows, _row: _remove_cache_after_update(db, rows, tablename)) + table._before_delete.append(lambda s: _remove_cache(db, s, tablename)) return cls diff --git a/tests/test_caching_multi_db.py b/tests/test_caching_multi_db.py new file mode 100644 index 0000000..0656b81 --- /dev/null +++ b/tests/test_caching_multi_db.py @@ -0,0 +1,88 @@ +""" +Two caching-enabled databases in one process must each keep their own cache (issue #21). +""" + +import pytest + +from src.typedal import TypeDAL, TypedTable +from src.typedal.caching import _TypedalCache, cache_models, clear_cache + + +# module level, because the cache stores rows with `dill`: +class MultiDbAppItem(TypedTable): + name: str + + +class MultiDbReportItem(TypedTable): + name: str + + +@pytest.fixture +def two_dbs(tmp_path): + app = TypeDAL(f"sqlite://{tmp_path / 'app.db'}", folder=str(tmp_path)) + report = TypeDAL(f"sqlite://{tmp_path / 'report.db'}", folder=str(tmp_path)) + app.define(MultiDbAppItem) + report.define(MultiDbReportItem) + yield app, report + report.close() + app.close() + + +def test_cache_models_are_per_database(two_dbs): + app, report = two_dbs + app_cache, app_dependency = cache_models(app) + report_cache, report_dependency = cache_models(report) + + assert app_cache is not report_cache + assert app_dependency is not report_dependency + assert app_cache._db is app + assert report_cache._db is report + # the module-level base class is never bound by enabling caching: + assert _TypedalCache._db is None + assert str(app_cache._table) == str(report_cache._table) == "typedal_cache" + + +def test_cache_rows_land_in_the_queried_database(two_dbs): + app, report = two_dbs + app_cache, _ = cache_models(app) + report_cache, _ = cache_models(report) + + MultiDbAppItem.insert(name="a") + assert MultiDbAppItem.cache().collect().metadata["cache"]["status"] == "fresh" + assert MultiDbAppItem.cache().collect().metadata["cache"]["status"] == "cached" + + assert app_cache.count() == 1 + assert report_cache.count() == 0 + + # invalidation through the insert hook touches only the app database: + MultiDbReportItem.insert(name="r") + MultiDbReportItem.cache().collect() + assert report_cache.count() == 1 + + MultiDbAppItem.insert(name="b") + assert app_cache.count() == 0 + assert report_cache.count() == 1 + + clear_cache(report) + assert report_cache.count() == 0 + + +def test_closing_one_database_keeps_caching_working_on_another(two_dbs, tmp_path): + app, _report = two_dbs + + other = TypeDAL(f"sqlite://{tmp_path / 'other.db'}", folder=str(tmp_path)) + other.close() + + # used to fail in the after-insert hook: "@define or db.define is not called on this class yet!" + MultiDbAppItem.insert(name="after close") + MultiDbAppItem.cache().collect() + assert cache_models(app)[0].count() == 1 + + +def test_cache_models_without_caching(tmp_path): + db = TypeDAL("sqlite:memory", enable_typedal_caching=False, folder=str(tmp_path)) + try: + with pytest.raises(RuntimeError): + cache_models(db) + finally: + db.close() diff --git a/tests/test_relationships.py b/tests/test_relationships.py index 12ab12d..397ca1a 100644 --- a/tests/test_relationships.py +++ b/tests/test_relationships.py @@ -10,8 +10,7 @@ from src.typedal import Relationship, TypeDAL, TypedField, TypedRows, TypedTable, relationship from src.typedal.caching import ( - _TypedalCache, - _TypedalCacheDependency, + cache_models, clear_cache, clear_expired, remove_cache, @@ -625,6 +624,7 @@ class Child(TypedTable): def test_caching(): _setup_data() + _TypedalCache, _TypedalCacheDependency = cache_models(db) uncached = User.join().collect_or_fail() cached = User.cache().join().collect_or_fail() # not actually cached yet! @@ -711,8 +711,8 @@ def test_caching(): assert User.cache().join().paginate(limit=1, page=2).metadata["cache"].get("status") == "cached" assert User.cache().join().paginate(limit=1, page=2).metadata["cache"].get("cached_at") - remove_cache(1, "user") - remove_cache([2], "user") + remove_cache(db, 1, "user") + remove_cache(db, [2], "user") assert User.cache("id").join().paginate(limit=1, page=1).metadata["cache"].get("status") == "fresh" assert User.cache().join().paginate(limit=1, page=2).metadata["cache"].get("status") == "fresh" @@ -727,7 +727,7 @@ def test_caching(): assert _TypedalCache.count() > 0 assert _TypedalCacheDependency.count() > 0 - clear_cache() + clear_cache(db) assert _TypedalCache.count() == 0 assert _TypedalCacheDependency.count() == 0 @@ -748,8 +748,8 @@ def test_caching(): time.sleep(3) # for TTL - assert clear_expired() - assert not clear_expired() + assert clear_expired(db) + assert not clear_expired(db) assert not _TypedalCache.count() assert not _TypedalCacheDependency.count() diff --git a/tests/test_upsert.py b/tests/test_upsert.py index d0cbf9d..3e69ebe 100644 --- a/tests/test_upsert.py +++ b/tests/test_upsert.py @@ -929,11 +929,7 @@ class CachedUpsert(TypedTable): @pytest.mark.parametrize("upsert_db", ["sqlite", "postgres", "mysql"], indirect=True) def test_upsert_invalidates_cache(upsert_db: RecordingDAL, tmp_path: Path) -> None: - from src.typedal.caching import _TypedalCache, _TypedalCacheDependency - - # the shared fixture disables caching; this one needs it. The cache models are module-global, so - # remember which database they belong to and hand them back afterwards. - previous = _TypedalCache._db + # the shared fixture disables caching; this one needs it uri = str(upsert_db._uri) db = TypeDAL(uri, enable_typedal_caching=True, folder=str(tmp_path / "cached")) try: @@ -966,9 +962,6 @@ def test_upsert_invalidates_cache(upsert_db: RecordingDAL, tmp_path: Path) -> No db.commit() db.close() CachedUpsert.unbind() - if previous is not None and previous._adapter is not None: - previous.try_define(_TypedalCache) - previous.try_define(_TypedalCacheDependency) @pytest.mark.parametrize("upsert_db", ["sqlite"], indirect=True)