Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions tests/test_tinydb.py
Original file line number Diff line number Diff line change
Expand Up @@ -677,6 +677,27 @@ def test_query_cache():
assert db.search(query) == [{'name': 'foo', 'value': 42}]


def test_query_cache_not_corrupted_by_mutation():
"""Mutating nested values of a search result must not pollute the cache."""
db = TinyDB(storage=MemoryStorage)
db.insert({'name': 'Alice', 'tags': ['a', 'b']})

query = where('name') == 'Alice'

results1 = db.search(query)
assert results1 == [{'name': 'Alice', 'tags': ['a', 'b']}]

# Mutate both a top-level value and a nested mutable inside the document
results1[0]['name'] = 'MUTATED'
results1[0]['tags'].append('c')

# A subsequent search hits the cache; it must return the original data
results2 = db.search(query)
assert results2 == [{'name': 'Alice', 'tags': ['a', 'b']}], (
'query cache was corrupted by mutation of a previously returned document'
)


def test_tinydb_is_iterable(db: TinyDB):
assert [r for r in db] == db.all()

Expand Down
8 changes: 5 additions & 3 deletions tinydb/table.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
"""

from collections.abc import Callable, Iterable, Iterator, Mapping
from copy import deepcopy
from typing import (
NoReturn,
Optional,
Expand Down Expand Up @@ -244,7 +245,7 @@ def search(self, cond: QueryLike) -> list[Document]:
# query
cached_results = self._query_cache.get(cond)
if cached_results is not None:
return cached_results[:]
return deepcopy(cached_results)

# Perform the search by applying the query to all documents.
# Then, only if the document matches the query, convert it
Expand All @@ -271,8 +272,9 @@ def search(self, cond: QueryLike) -> list[Document]:
is_cacheable: Callable[[], bool] = getattr(cond, 'is_cacheable',
lambda: True)
if is_cacheable():
# Update the query cache
self._query_cache[cond] = docs[:]
# Update the query cache with a deep copy so that callers
# mutating returned documents cannot corrupt cached results.
self._query_cache[cond] = deepcopy(docs)

return docs

Expand Down