Repository navigation
Validation function of st.cache_resource() to receive the same arguments that are passed to the cached function. #14357
Description
Checklist
- I have searched the existing issues for similar feature requests.
- I added a descriptive title and summary to this issue.
Summary
As seen here in the documentation of st.cache_resource(), validate
receives the cached value as its only parameter and it must return a boolean.
I'd like the validation function to receive the same arguments that are passed to the cached function.
Why?
As seen here in the documentation, validate
is useful e.g. to check the health of database connections.
The cached function could be responsible for caching multiple database connections, based on an argument it receives. The validation function would then need to receive the same argument(s) to check the health of the corresponding database connection.
As a simpler, more concrete example, I'd like to cache the content of big files and invalidate the cache if the files change. Note the addition of the path argument in validate:
@st.cache_resource(validate=lambda cached, path: cached[0] == path.stat().st_mtime)
def get_big_file(path: Path):
return path.stat().st_mtime, load_big_file_with_expensive_compute(path)Side note: depending on the situation, st.cache_data() might be more appropriate than @st.cache_resource(), so the validate argument could also be added to st.cache_data().
How?
I can think of two ways to avoid a breaking change to the current validate function:
- We could inspect if
validateexpects arguments, something likevalidate(*args, *kwargs) if inspect.signature(validate).parameters else validate()
- We could introduce a new function like
validate_with_args. The two functions would be mutually exclusive.
Additional Context
As a workaround, I can create one cached function for every file:
# File 1
@st.cache_resource(validate=lambda cached: cached[0] == path1.stat().st_mtime)
def get_big_file1():
return path1.stat().st_mtime, load_big_file_with_expensive_compute(path)
# File 2
@st.cache_resource(validate=lambda cached: cached[0] == path2.stat().st_mtime)
def get_big_file2():
return path2.stat().st_mtime, load_big_file_with_expensive_compute(path)But if I my paths are dynamic, that gets complicated. I implemented a decorator that dynamically creates a cached function (with a unique __qualname__) as well as the corresponding validation function for each path. See cache_file below.
Without the __qualname__, the first file would work, but not any subsequent file, as the first validate function would be used for all the files. That's feels hacky/fragile.
from collections.abc import Callable
from pathlib import Path
from typing import TypeVar
from unittest.mock import MagicMock, patch
import streamlit as st
from streamlit.testing.v1 import AppTest
T = TypeVar("T")
def cache_file(f: Callable[[Path], T]) -> Callable[[Path], T]:
def decorated(path: Path, **kwargs):
def validate(cached):
return cached[0] == path.stat().st_mtime
def cached(path: Path, **kwargs):
return path.stat().st_mtime, f(path, **kwargs)
cached.__qualname__ = str(path)
return st.cache_resource(cached, validate=validate)(path, **kwargs)[1]
return decorated
def _test_app(path0, path1):
from pathlib import Path
import streamlit as st
from apps.utils import cache_file
@cache_file
def get_text(path: Path):
return path.read_text()
st.text(get_text(path0))
st.text(get_text(path1))
@patch.object(Path, "read_text", autospec=True, wraps=Path.read_text)
def test_cache_file(mock_read_text: MagicMock, tmp_path: Path):
paths = (tmp_path / "0.txt", tmp_path / "1.txt")
paths[0].write_text("0A")
paths[1].write_text("1A")
at = AppTest.from_function(_test_app, args=paths)
for _ in range(2):
at.run()
assert not at.exception
assert at.text[0].value == "0A"
assert at.text[1].value == "1A"
assert mock_read_text.call_count == 2
paths[0].write_text("0B")
for _ in range(2):
at.run()
assert not at.exception
assert at.text[0].value == "0B"
assert at.text[1].value == "1A"
assert mock_read_text.call_count == 3
paths[0].touch()
for _ in range(2):
at.run()
assert not at.exception
assert at.text[0].value == "0B"
assert at.text[1].value == "1A"
assert mock_read_text.call_count == 4
paths[1].write_text("1B")
for _ in range(2):
at.run()
assert not at.exception
assert at.text[0].value == "0B"
assert at.text[1].value == "1B"
assert mock_read_text.call_count == 5
Activity
github-actions commentedon Mar 12, 2026
To help Streamlit prioritize this feature, react with a 👍 (thumbs up emoji) to the initial post.
Your vote helps us identify which enhancements matter most to our users.