import threading from datetime import timedelta from requests_cache import CachedSession # Cache for "semi-static" data retrieved from a URL. This is a layet of caching in addition to the normal caching of # spots, alerts and other data in the DataStore class. Its purpose is to avoid hitting remote endpoints frequently when # e.g. restarting Spothole many times during testing. _session = CachedSession("cache/semi_static_urls", expire_after=timedelta(days=1), allowable_codes=(200, 400, 401, 403, 404)) _lock = threading.Lock() class _ThreadSafeSession: """Wraps CachedSession with a lock to prevent concurrent SQLite access across threads. This allows a single object to be used freely across the application.""" def get(self, *args, **kwargs): with _lock: return _session.get(*args, **kwargs) # Global object URL_DATA_CACHE = _ThreadSafeSession()